Work队列消息模型

Work队列消息模型，也被称作工作队列消息模型，或者竞争消费模型。

Work队列消息模型与基本消息队列模型的区别就在于，Work队列消息模型让多个消费者绑定一个队列，这样就可以快速处理消息队列中的消息，从而避免了队列中消息堆积的问题。

JavaAPI实现Work队列消息模型

准备工作

使用JavaAPI实现消息队列模型，需要先做一些准备工作，这些准备工作包括搭建项目环境和准备工具类，这些工作在消息队列RabbitMQ之基本消息模型已经准备好了，就不重复准备了。

创建生产者

public class Send {
    //RabbitMQ的队列名称
    private final static String QUEUE_NAME = "work_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        //使用工具类，获取RabbitMQ的连接
        Connection connection = ConnectionUtils.getConnection();
        //使用连接创建通道
        Channel channel = connection.createChannel();
        //使用通道声明一个队列，参数分别为：队列名称，是否持久化，是否为排他队列，是否自动删除，其他属性
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //循环发送50条消息到消息队列中
        for(int i = 0; i < 50; i++){
            //需要发送的消息
            String message = "Hello World!" + i;
            //使用通道发送消息，参数分别为：交换机名称，路由Key，其他属性，消息正文
            channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
        }
        //关闭通道
        channel.close();
        //关闭连接
        connection.close();
    }
}

创建消费者1

public class Recv1 {
    //RabbitMQ的队列名称
    private final static String QUEUE_NAME = "work_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        //使用工具类，获取RabbitMQ的连接
        Connection connection = ConnectionUtils.getConnection();
        //使用连接创建通道
        Channel channel = connection.createChannel();
        //使用通道声明一个队列，参数分别为：队列名称，是否持久化，是否为排他队列，是否自动删除，其他属性
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //设置每个消费者每次只消费一条数据
        channel.basicQos(1);
        //定义队列的消费者
        DefaultConsumer consumer = new DefaultConsumer(channel){
            //覆写该方法，该方法类似于监听器，当队列中有消息的时候，就会被调用执行
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //byte[] body参数即为消息体，当该消息体被使用，即认为该消息已被消费
                System.out.println("[消费者1]"+new String(body));
                //手动Ack
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        //监听队列，参数分别为：队列名称，是否自动确认消息已被消费，队列消费者
        channel.basicConsume(QUEUE_NAME,false,consumer);
    }
}

创建消费者2

public class Recv2 {
	//RabbitMQ的队列名称
    private final static String QUEUE_NAME = "work_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        //使用工具类，获取RabbitMQ的连接
        Connection connection = ConnectionUtils.getConnection();
        //使用连接创建通道
        Channel channel = connection.createChannel();
        //使用通道声明一个队列，参数分别为：队列名称，是否持久化，是否为排他队列，是否自动删除，其他属性
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //设置每个消费者每次只消费一条数据
        channel.basicQos(1);
        //定义队列的消费者
        DefaultConsumer consumer = new DefaultConsumer(channel){
            //覆写该方法，该方法类似于监听器，当队列中有消息的时候，就会被调用执行
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //byte[] body参数即为消息体，当该消息体被使用，即认为该消息已被消费
                System.out.println("[消费者2]"+new String(body));
                //第二个消费者处理消息较慢
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        //监听队列，参数分别为：队列名称，是否自动确认消息已被消费，队列消费者
        channel.basicConsume(QUEUE_NAME,false,consumer);
    }
}

Work队列消息模型代码实现的注意点

与基本消息模型的实现相比，Work队列消息模型在代码实现上多了这么一行代码：

1 2	//设置每个消费者每次只消费一条数据 channel.basicQos(1);

这是将消费者设置为每次只消费一条数据，如果不设置这个值，那么在一开始，每个消费者都是默认平均消费的，即每个消费者各消费25条消息。
但是在消费者2中，消息的处理速度非常慢（每处理一条消息，线程休眠一秒钟），所以为了能够最大化利用资源，所以设置每个消费者每次只消费一条消息，这样消费消息更快的消费者就可以消费更多的消息。

【注意】在Work队列消息模型中，应该采用手动Ack的方式，因为如果是自动Ack，那么每次已使用body的消息体，消费者就自动Ack告诉RabbitMQ消息已被消费，那么RabbitMQ就会立刻将下一条消息发送给该消费者，这样消费者的线程休眠问题就体现不出来。