RabbitMQ整合Spring Booot【公平队列】

producer：

package com.toov5.Producer;

import java.io.IOException; import java.util.concurrent.TimeoutException; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.toov5.utils.MQConnectionUtils; public class Producer { // 队列名称 private static final String UEUE_NAME = "test_queue"; public static void main(String[] args) throws IOException, TimeoutException { // 建立新的链接 Connection connection = MQConnectionUtils.newConnection(); // 建立Channel Channel channel = connection.createChannel(); // 建立队列 channel.queueDeclare(UEUE_NAME, false, false, false, null);  channel.basicQos(1); // 保证 取一个消费 队列给消费者发送消息时候 一个消息 for (int i = 0; i < 10; i++) { // 建立message String msg = "toov5_message"; System.out.println("生产者投递消息" + msg + i); // 生产者发送消息 channel.basicPublish("", UEUE_NAME, null, msg.getBytes()); } // 关闭通道和链接  channel.close(); connection.close(); } }

 

Consumer1

package com.toov5.Consumer;

import java.io.IOException; import java.util.concurrent.TimeoutException; import com.rabbitmq.client.AMQP.BasicProperties; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.DefaultConsumer; import com.rabbitmq.client.Envelope; import com.toov5.utils.MQConnectionUtils; public class Consumer1 { //队列名称 private static final String QUEUE_NAME = "test_queue"; public static void main(String[] args) throws IOException, TimeoutException { System.out.println("消费者启动..........1"); //建立新的链接 Connection connection = MQConnectionUtils.newConnection(); //建立Channel final Channel channel = connection.createChannel(); // 消费者关联队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); channel.basicQos(1); DefaultConsumer defaultConsumerr = new DefaultConsumer(channel) { //监听获取消息  @Override public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException { String msg =new String(body,"UTF-8"); System.out.println("消费者获取生产者消息："+msg); try { //模拟应答等待时间 Thread.sleep(1000); } catch (Exception e) { }finally { channel.basicAck(envelope.getDeliveryTag(), false); //手动应答 告诉消息队列服务器 消费成功  } } }; //牵手模式设置 默认自动应答模式 true:自动应答模式 channel.basicConsume(QUEUE_NAME, false, defaultConsumerr);// fanse手动应答  } }

 

Consumer2

package com.toov5.Consumer;

import java.io.IOException; import java.util.concurrent.TimeoutException; import com.rabbitmq.client.AMQP.BasicProperties; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.DefaultConsumer; import com.rabbitmq.client.Envelope; import com.toov5.utils.MQConnectionUtils; public class Consumer2 { //队列名称 private static final String QUEUE_NAME = "test_queue"; public static void main(String[] args) throws IOException, TimeoutException { System.out.println("消费者启动..........2"); //建立新的链接 Connection connection = MQConnectionUtils.newConnection(); //建立Channel final Channel channel = connection.createChannel(); // 消费者关联队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); channel.basicQos(1); DefaultConsumer defaultConsumerr = new DefaultConsumer(channel) { //监听获取消息  @Override public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException { String msg =new String(body,"UTF-8"); System.out.println("消费者获取生产者消息："+msg); try { //模拟应答等待时间 Thread.sleep(300); } catch (Exception e) { }finally { channel.basicAck(envelope.getDeliveryTag(), false); //手动应答 告诉消息队列服务器 消费成功  } } }; //牵手模式设置 默认自动应答模式 true:自动应答模式 channel.basicConsume(QUEUE_NAME, false, defaultConsumerr);// fanse手动应答  } }

运行结果：

睡眠少的（执行快的） 指定的多

 注意 每一个消费者 必需要应答 一下！ 队列服务器没有收到应答 就不会发送下一个给消费者~