首页 \ 教程 \ Java

RabbitMQ 教程

RabbitMQ进阶教程

RabbitMQ入门教程

开源项目

Jersey

知识点

Java

RabbitMQ 简介

RabbitMQ 简单队列

RabbitMQ 消息确认方式

RabbitMQ topic Exchange使用

RabbitMQ 与Spring整合

RabbitMQ 发布订阅模式(Publish、Subscribe)

RabbitMQ 路由模式（Routing）-使用 direct Exchange

RabbitMQ 三种Exchange(fanout,direct,topic)模式

RabbitMQ 插件管理

RabbitMQ 集群搭建

RabbitMQ 环境配置-基于linux

RabbitMQ vhost管理

RabbitMQ 用户管理

RabbitMQ 环境配置-基于windows 10 x64

ActiveMQ中queues.jsp显示的队列信息

RabbitMQ Work模式消息队列

2019-03-05 22:54|来源: 网路

一个生产者、多个消费者。一个消息只能被一个消费者获取。

生产者发布消息

private final static String QUEUE_NAME = "test_queue_work";
@Test
public void testSend() throws Exception {
    // 获取到连接以及mq通道
    Connection connection = ConnectionUtil.getConnection();
    Channel channel = connection.createChannel();
    // 声明队列
    channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    for (int i = 0; i < 50; i++) {
        // 消息内容
        String message = "" + i;
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
        System.out.println(" [x] Sent '" + message + "'");
        Thread.sleep(i * 10);
    }
    channel.close();
    connection.close();
}

消息者1

@Test
public void testRecv1() throws Exception {
    // 获取到连接以及mq通道
    Connection connection = ConnectionUtil.getConnection();
    Channel channel = connection.createChannel();
    // 声明队列
    channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    // 定义队列的消费者
    QueueingConsumer consumer = new QueueingConsumer(channel);
    // 监听队列，手动返回完成
    channel.basicConsume(QUEUE_NAME, false, consumer);
    // 获取消息
    while (true) {
        QueueingConsumer.Delivery delivery = consumer.nextDelivery();
        String message = new String(delivery.getBody());
        System.out.println(" [x] Received '" + message + "'");
        //休眠
        Thread.sleep(10);
        // 返回确认状态
        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    }
}

消费者2

@Test
public void testRecv2() throws Exception{
    // 获取到连接以及mq通道
    Connection connection = ConnectionUtil.getConnection();
    Channel channel = connection.createChannel();
    // 声明队列
    channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    // 定义队列的消费者
    QueueingConsumer consumer = new QueueingConsumer(channel);
    // 监听队列，手动返回完成状态
    channel.basicConsume(QUEUE_NAME, false, consumer);
    // 获取消息
    while (true) {
        QueueingConsumer.Delivery delivery = consumer.nextDelivery();
        String message = new String(delivery.getBody());
        System.out.println(" [x] Received '" + message + "'");
        // 休眠1秒
        Thread.sleep(1000);
        //反馈消息的消费状态
        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    }
}

为了测试消费者获取到的消息情况，先将两个消费者启动后，再启动生产者

消费者1打印的消息

消费者2打印的消息

[x] Received '0'

[x] Received '2'

[x] Received '4'

[x] Received '6'

[x] Received '8'

[x] Received '10'

[x] Received '12'

[x] Received '14'

[x] Received '16'

[x] Received '18'

[x] Received '20'

[x] Received '22'

[x] Received '24'

[x] Received '26'

[x] Received '28'

[x] Received '30'

[x] Received '32'

[x] Received '34'

[x] Received '36'

[x] Received '38'

[x] Received '40'

[x] Received '42'

[x] Received '44'

[x] Received '46'

[x] Received '48'

[x] Received '1'

[x] Received '3'

[x] Received '5'

[x] Received '7'

[x] Received '9'

[x] Received '11'

[x] Received '13'

[x] Received '15'

[x] Received '17'

[x] Received '19'

[x] Received '21'

[x] Received '23'

[x] Received '25'

[x] Received '27'

[x] Received '29'

[x] Received '31'

[x] Received '33'

[x] Received '35'

[x] Received '37'

[x] Received '39'

[x] Received '41'

[x] Received '43'

[x] Received '45'

[x] Received '47'

[x] Received '49'

测试结果：

消费者1和消费者2获取到的消息内容是不同的，同一个消息只能被一个消费者获取。

消费者1和消费者2获取到的消息的数量是相同的，一个是奇数一个是偶数。

但是由于消费者1每次消费后只休息10毫秒，而消费者2每次消费后只休息1秒，这样是不合理的，应该是消费者1要比消费者2获取到的消息多才对。

此时可以使用channel.basicQos(1);来设置同一时刻服务器只会发一条消息给消费者，解决“能者多劳”问题，以消费者1为例：

@Test
public void testRecv1() throws Exception {
    // 获取到连接以及mq通道
    Connection connection = ConnectionUtil.getConnection();
    Channel channel = connection.createChannel();
    // 声明队列
    channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    // 同一时刻服务器只会发一条消息给消费者
    channel.basicQos(1);
    // 定义队列的消费者
    QueueingConsumer consumer = new QueueingConsumer(channel);
    // 监听队列，手动返回完成
    channel.basicConsume(QUEUE_NAME, false, consumer);
    // 获取消息
    while (true) {
        QueueingConsumer.Delivery delivery = consumer.nextDelivery();
        String message = new String(delivery.getBody());
        System.out.println(" [x] Received '" + message + "'");
        //休眠
        Thread.sleep(10);
        // 返回确认状态
        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    }
}

执行结果：