1.介绍

简单队列有个缺点，简单队列是一一对应的关系，即点对点，一个生产者对应一个消费者，按照这个逻辑，如果我们有一些比较耗时的任务，也就意味着需要大量的时间才能处理完毕，显然简单队列模式并不能满足我们的工作需求，我们今天再来看看工作队列。

一个生产者对应多个消费者，但是只能有一个消费者获得消息！！！

创建maven项目，导入依赖

        <dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.7.3</version></dependency><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId><version>1.7.25</version><scope>test</scope></dependency><dependency><groupId>log4j</groupId><artifactId>log4j</artifactId><version>1.2.17</version></dependency><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>4.12</version></dependency>

2.消息生产者

发送50条消息

/*** 消息生产者* @Author liuqingsen* @Date 2020/6/21 15:10*/
public class Producer {//队列名private static final String QUEUE_NAME = "work_queue";public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {//获取连接Connection connection = ConnectionUtil.getConnection();//创建通道Channel channel = connection.createChannel();//声明队列channel.queueDeclare(QUEUE_NAME,false,false,false,null);for ( int i = 0 ; i < 50 ; i++){String msg = "生产者消息=>"+i;System.out.println("生产者发送消息:" + msg);//发送消息channel.basicPublish("",QUEUE_NAME,null,msg.getBytes()  );//模拟发送消息延时，便于演示多个消费者竞争接受消息Thread.sleep(i*10);}channel.close();connection.close();}
}

3.消息消费者

消费者1，等待10毫秒，模拟处理详细效率 快

/*** 消费者1* @Author liuqingsen* @Date 2020/6/21 15:19*/
public class Customer_1 {//队列名private static final String QUEUE_NAME = "work_queue";public static void main(String[] args) throws IOException, TimeoutException {//获取连接Connection connection = ConnectionUtil.getConnection();//创建通道Channel channel = connection.createChannel();//声明队列channel.queueDeclare(QUEUE_NAME,false,false,false,null);//定义一个消费者DefaultConsumer defaultConsumer = new DefaultConsumer(channel){//一旦该队列有消息就会触发@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String msg = new String(body,"utf-8");System.out.println("消费者[1]获取消息:" + msg);try {Thread.sleep(10);} catch (InterruptedException e) {e.printStackTrace();}finally {//返回确认状态channel.basicAck(envelope.getDeliveryTag(), false);}}};//监听队列channel.basicConsume(QUEUE_NAME,false,defaultConsumer);}
}

消费者2，等待时间3000毫秒，模拟处理消息效率 慢

/*** 消费者2* @Author liuqingsen* @Date 2020/6/21 15:39*/
public class Customer_2 {//队列名private static final String QUEUE_NAME = "work_queue";public static void main(String[] args) throws IOException, TimeoutException {//获取连接Connection connection = ConnectionUtil.getConnection();//创建通道Channel channel = connection.createChannel();//声明队列channel.queueDeclare(QUEUE_NAME,false,false,false,null);//定义一个消费者DefaultConsumer defaultConsumer = new DefaultConsumer(channel){//一旦该队列有消息就会触发@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String msg = new String(body,"utf-8");System.out.println("消费者[2]获取消息:" + msg);try {Thread.sleep(3000);} catch (InterruptedException e) {e.printStackTrace();}finally {//返回确认状态channel.basicAck(envelope.getDeliveryTag(), false);}}};//监听队列channel.basicConsume(QUEUE_NAME,false,defaultConsumer);}
}

生产者控制台打印：

消费者1控制台打印：

消费者2控制台打印：

可以看出是轮询分发消息的

存在的问题：消费者处理能力没有分配好，我把消息分发给你们，你们什么时候处理完，不管我事！
消费者1比消费者2执行效率更高，分配了相同数量的消息，这样显然是不合理的！

4.能者多劳

其实发生上述问题的原因是 RabbitMQ 收到消息后就立即分发出去，而没有确认各个工作者未返回确认的消息数量，类似于TCP/UDP中的UDP，面向无连接。

因此我们可以使用 basicQos 方法，并将参数 prefetchCount 设为1，告诉 RabbitMQ 我每次值处理一条消息，你要等我处理完了再分给我下一个。这样 RabbitMQ 就不会轮流分发了，而是寻找空闲的工作者进行分发。

关键性代码：

//声明队列
channel.queueDeclare(QUEUE_NAME,false,false,false,null);
//保证一次只分发一次 限制发送给同一个消费者 不得超过一条消息
channel.basicQos(1);

消费者1控制台打印

消费者2控制台打印

可以明显看出，执行效率高的（消费者1）分配的消息更多，执行效率低的分配的消息更少。

5.消息持久化

当有多个消费者同时收取消息，且每个消费者在接收消息的同时，还要处理其它的事情，且会消耗很长的时间。在此过程中可能会出现一些意外，比如消息接收到一半的时候，一个消费者死掉了。

这种情况要使用消息接收确认机制，可以执行上次宕机的消费者没有完成的事情。

但是在默认情况下，我们程序创建的消息队列以及存放在队列里面的消息，都是非持久化的。当RabbitMQ死掉了或者重启了，上次创建的队列、消息都不会保存。

参数配置:
参数配置一：生产者创建队列声明时，修改第二个参数为 true

/**3.创建队列声明 */
channel.queueDeclare(QUEUE_NAME, true, false, false, null);

参数配置二：生产者发送消息时，修改第三个参数为MessageProperties.PERSISTENT_TEXT_PLAIN

for (int i = 1; i <= 50; i++) {String msg = "生产者消息_" + i;System.out.println("生产者发送消息:" + msg);/**4.发送消息 */channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes());
}