概述

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

以上的角色分别为如下所解释的：

• P：生产者：任务的发布者
• C1：消费者1，领取任务并且完成任务，假设完成速度较慢
• C2：消费者2：领取任务并完成任务，假设完成速度较快

创建生产者

代码如下所示：

/**
 * @author BNZeng
 */
public class Consumer1 {

    @Test
    public void receiveMessage() throws Exception {

        Connection connection = RabbitMQUtil.getConnection();

        Channel channel = connection.createChannel();

        channel.queueDeclare("hello", false, false, false, null);
        channel.basicConsume("hello", true, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者【1】收到消息 → " + new String(body));
            }
        });

        System.out.println("消费者【1】启动成功");

        // 不能让程序结束
        System.in.read();

        // 释放资源
        RabbitMQUtil.closeChannelAndConnection(channel, connection);
    }
}

创建消费者 2

代码如下所示：

/**
 * @author BNTang
 */
public class Consumer2 {

    @Test
    public void receiveMessage() throws Exception {

        Connection connection = RabbitMQUtil.getConnection();

        Channel channel = connection.createChannel();

        // 一次只处理一条消息
        channel.basicQos(1);
        channel.queueDeclare("hello", false, false, false, null);

        // 把签收模式变成 false
        channel.basicConsume("hello", false, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                // 手动签收
                channel.basicAck(envelope.getDeliveryTag(), false);
                System.out.println("消费者【2】收到消息 → " + new String(body));
            }
        });

        System.out.println("消费者【2】启动成功");

        // 不能让程序结束
        System.in.read();

        // 释放资源
        RabbitMQUtil.closeChannelAndConnection(channel, connection);
    }
}

运行起来进行测试，结果如下所示：