RabbitMQ如何高效的消费消息

在上篇介绍了如何简单的发送一个消息队列以后，咱们本篇来看下RabbitMQ的另一种模式，工做队列。ide

什么是工做队列

咱们上篇文章说的是，一个生产者生产了消息被一个消费者消费了，以下图学习

上面这种简单的消息队列确实能够处理咱们的任务，可是当咱们队列中的任务过多，处理每条任务有须要很长的耗时，那么使用一个消费者处理消息显然不不够的，因此咱们能够增长消费者，来共享消息队列中的消息，进行任务处理。spa

也就是以下图code

虽然上图我只花了一个生产者A，那么同理，能有多个消费者，那也能多个生产者。队列

代码

发送消息

public class Send {

    public static final String QUEUE_NAME = "test_word_queue";

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {

        // 获取链接
        Connection connection = MQConnectUtil.getConnection();

        // 建立通道
        Channel channel = connection.createChannel();

        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 模拟发送20条消息
        for (int i = 0; i < 20; i++) {

            String msg = "消息：" + i;

            channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());

            Thread.sleep(i * 20);

            System.out.println(msg);
        }

        channel.close();
        connection.close();
    }
}

消费者A

public class Consumer1 {

    public static final String QUEUE_NAME = "test_word_queue";

    public static void main(String[] args) throws Exception {

        // 获取链接
        Connection connection = MQConnectUtil.getConnection();

        // 建立频道
        Channel channel = connection.createChannel();

        // 队列声明
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 定义消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @SneakyThrows
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                String msg = new String(body, StandardCharsets.UTF_8);

                System.out.println("消费者[A]-内容：" + msg);

                Thread.sleep(2 * 1000);
            }
        };

        // 监听队列
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

消费者B

public class Consumer2 {

    public static final String QUEUE_NAME = "test_word_queue";

    public static void main(String[] args) throws Exception {

        // 获取链接
        Connection connection = MQConnectUtil.getConnection();

        // 建立频道
        Channel channel = connection.createChannel();

        // 队列声明
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 定义消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @SneakyThrows
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                String msg = new String(body, StandardCharsets.UTF_8);

                System.out.println("消费者[B]-内容：" + msg);

                Thread.sleep(1000);
            }
        };

        // 监听队列
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

咱们来看下消费者A和消费者B的消费状况rem

• 消费者B

• 消费者A

有没有发现什么问题，我总过模拟发送了20条消息，细心的同窗能够发现，消费者A和消费者B消费了一样多的消息，都消费了10天，可是我在消费者A和消费者B中,什么sleep不通的时长,按道理说消费者B要比消费者A处理消息的速度块，处理的消息更多，那么为何会产生这样的缘由？get

RabbitMQ工做队列的默认配置

默认状况下，RabbitMQ会将每一个消息依次发送给下一个消费者，每一个消费者收到的消息数量实际上是同样的，咱们把这种分发消息的方式称为轮训分发模式。消息队列

本篇咱们就简单介绍这么多内容，有心学习的童鞋必定要敲敲代码，看不必定能看会，只有本身敲一遍，才能有所理解。it

更多内容请关注：io