WorkQueues，也被称为任务队列模型。当消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度。长此以往，消息就会堆积越来越多，无法及时的处理。此时就可以使用work模型：让多个消费者绑定到一个队列，共同消费队列中的消息。队列中的消息一旦消费，就会消失，因为任务是不会被重复执行的。

P：生产者

C1：消费者-1，领取任务并且完成任务，假设完成速度较慢。

C2：消费者-2，领取任务并且完成任务，假设完成速度快。

1.生产者

public class Provider {
    //生产消息
    @Test
    public void testSendMessage() throws IOException, TimeoutException {
        Connection connection = RabbitMqUtil.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare("work",true,false,false,null);
        for(int i = 1;i<=200;i++){
            channel.basicPublish("","work",null,(i+"hello rabbitmq").getBytes());
        }
        RabbitMqUtil.closeConnectionAndChannel(channel,connection);
    }
}

2.消费者-1

 

public class Consumer1 {
    //消费消息，这里需要用main函数，因为消费端要一直监听队列，而test测试会直接结束
    public static void main(String[] args) throws IOException, TimeoutException {
        //获取连接对象
        Connection connection = RabbitMqUtil.getConnection();
        //获取连接通道
        Channel channel = connection.createChannel();
        channel.queueDeclare("work",true,false,false,null);
        channel.basicConsume("work",true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                try {
                    Thread.sleep(2000);
                    System.out.println("consumer1得到："+new String(body));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        //注意这里不能关闭通道和连接，因为要一直监听
    }
}

 3.消费者-2

public class Consumer2 {
    //消费消息，这里需要用main函数，因为消费端要一直监听队列，而test测试会直接结束
    public static void main(String[] args) throws IOException, TimeoutException {
        //获取连接对象
        Connection connection = RabbitMqUtil.getConnection();
        //获取连接通道
        Channel channel = connection.createChannel();
        channel.queueDeclare("work",true,false,false,null);
        channel.basicConsume("work",true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("consumer2得到："+new String(body));
            }
        });
        //注意这里不能关闭通道和连接，因为要一直监听
    }
}

4.结果分析

通过运行结果我们发现消费者1和消费2是平均处理消息的，就比如1000个消息一人处理一半。而且现在的机制是，队列中的消息会一次性全部分配给两个消费者，囤积到两个消费者处然后让两个消费者去各自慢慢消化。这样就会产生一些问题：

1.由于我们设置了消息的自动确认机制，两个消费者刚得到大量消息都还没开始消费其实就已经告诉队列我们确认完了，这样显然是不合理的。

2.消费者那边一次性囤积了大量未处理的消息，如果处理中宕机了，囤积的消息会丢失。

而且假如消费者2执行的很快，而另一个消费者1执行的很慢，这样消费者2很快执行完就空闲了，而消费者1一直迟迟执行不完。能不能改进为能者多劳的机制呢？

• 消费者一次只接收一条未确认的消息。

• 关闭自动确认消息。

• 消费者处理完一条，要手动确认消息。

5.能者多劳

消费者改进：

public class Consumer1Improve {
    //消费消息，这里需要用main函数，因为消费端要一直监听队列，而test测试会直接结束
    public static void main(String[] args) throws IOException, TimeoutException {
        //获取连接对象
        Connection connection = RabbitMqUtil.getConnection();
        //获取连接通道
        Channel channel = connection.createChannel();
        channel.queueDeclare("work",true,false,false,null);
        channel.basicQos(1);//一次只接收一条
        //参数2：关闭自动确认
        channel.basicConsume("work",false,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                try {
                    Thread.sleep(2000);
                    System.out.println("consumer1得到："+new String(body));
                    channel.basicAck(envelope.getDeliveryTag(),false);//手动确认消息
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        //注意这里不能关闭通道和连接，因为要一直监听
    }
}