AMQP-RabbitMQ/2/工作队列

2019-06-08 08:52:45 阅读：238 来源： 互联网

标签：false AMQP 队列 futao RabbitMQ 消息 import com channel

2. 工作队列 Work queues

Distributing tasks among workers

消息将发送给c1或者c2

# 个人理解

生产者定义Queue，并向该队列发送消息
多个消费者可以从指定的同一个Queue中读取消息。每条消息只会发送给其中某一个消费者。
生产者

package com.futao.springmvcdemo.mq.rabbit.workqueue;

import com.futao.springmvcdemo.mq.rabbit.RabbitMqConnectionTools;
import com.futao.springmvcdemo.mq.rabbit.RabbitMqQueueEnum;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import lombok.Cleanup;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

/**
 * 简单发送者
 *
 * @author futao
 * Created on 2019-04-22.
 */
@Slf4j
public class Send {
    @SneakyThrows
    public static void main(String[] args) {
        @Cleanup
        Connection connection = RabbitMqConnectionTools.getConnection();
        @Cleanup
        Channel channel = connection.createChannel();
        //开启持久化消息
        boolean durable = true;
        //定义一个队列
        channel.queueDeclare(RabbitMqQueueEnum.WORK_QUEUE.getQueueName(), durable, false, false, null);
        String msg = "Hello RabbitMq!";
        for (int i = 0; i < 20; i++) {
            channel.basicPublish("", RabbitMqQueueEnum.WORK_QUEUE.getQueueName(), null, (msg + i).getBytes());
            log.info("Send msg:[{}] success", (msg + i));
        }
    }
}

消费者1 - 每1秒处理一条

package com.futao.springmvcdemo.mq.rabbit.workqueue;

import com.futao.springmvcdemo.mq.rabbit.RabbitMqConnectionTools;
import com.futao.springmvcdemo.mq.rabbit.RabbitMqQueueEnum;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

/**
 * 简单消费者
 *
 * @author futao
 * Created on 2019-04-22.
 */
@Slf4j
public class RecvOne {
    @SneakyThrows
    public static void main(String[] args) {
        Channel channel = RabbitMqConnectionTools.getChannel();
        //开启持久化消息
        boolean durable = true;
        channel.queueDeclare(RabbitMqQueueEnum.WORK_QUEUE.getQueueName(), durable, false, false, null);
        log.info("Waiting for message...");
        DeliverCallback deliverCallback = ((consumerTag, message) -> {
            log.info("收到消息:[{}],tag:[{}]", new String(message.getBody()), consumerTag);
            //acknowledgment应答
            channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
            try {
                Thread.sleep(1000);
            } catch (Exception e) {

            }
        });
        //关闭自动应答
        boolean autoAck = false;
        channel.basicConsume(RabbitMqQueueEnum.WORK_QUEUE.getQueueName(), autoAck, deliverCallback, consumerTag -> {
        });
    }
}

消费者2 - 每2秒处理一条

package com.futao.springmvcdemo.mq.rabbit.workqueue;

import com.futao.springmvcdemo.mq.rabbit.RabbitMqConnectionTools;
import com.futao.springmvcdemo.mq.rabbit.RabbitMqQueueEnum;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

/**
 * 简单消费者
 *
 * @author futao
 * Created on 2019-04-22.
 */
@Slf4j
public class RecvTwo {
    @SneakyThrows
    public static void main(String[] args) {
        Channel channel = RabbitMqConnectionTools.getChannel();
        //开启持久化消息
        boolean durable = true;
        channel.queueDeclare(RabbitMqQueueEnum.WORK_QUEUE.getQueueName(), durable, false, false, null);
        log.info("Waiting for message...");
        DeliverCallback deliverCallback = ((consumerTag, message) -> {
            log.info("收到消息:[{}],tag:[{}]", new String(message.getBody()), consumerTag);
            //acknowledgment应答
            channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
            try {
                Thread.sleep(2000);
            } catch (Exception e) {

            }
        });
        //关闭自动应答
        boolean autoAck = false;
        channel.basicConsume(RabbitMqQueueEnum.WORK_QUEUE.getQueueName(), autoAck, deliverCallback, consumerTag -> {
        });
    }
}

结果
- 生产者
- 消费者1日志
- 消费者2日志
特点: 多个消费者之间，不论消息的处理速度，都是平均分发(公平分发)。你一个，我一个，他一个。此时是公平队列
注意：
- 定义队列的时候，设置是否开启消息的持久化（该设置需要同时在生产者和消费者设置）

       //开启持久化消息
        boolean durable = true;
        channel.queueDeclare(RabbitMqQueueEnum.WORK_QUEUE.getQueueName(), durable, false, false, null);

如果消息队列已经存在，则不可以修改相应的配置，必须删除原有的队列，或者新建一个新的队列。
关闭自动应答（开启手动应答），可以防止消息在未被正确消费的情况下被Rabbitmq从队列内存中删除。

# 实现工作队列下的非公平队列

消费者设置一次只发送一条消息，并且在被正确消费之前发继续发送下一条消息。从而使得消费快的消费者比消费慢的消费者消费更多的消息

        //告诉rabbitmq一次只发送一条消息，并且在前一个消息未被处理或者消费之前，不继续发送下一个消息
        channel.basicQos(1);

测试结果

此时明显打破了消息的公平分发，消费快的消费者接收到的消息更多。
如果有两个消费者，其中一个ConsumerA设置了Qos=1，另一个ConsumerB没有设置。经过我的测试，ConsumerA会获得大量的消息，都积压在ConsumerA，而ConsumerB获得消息很少。

标签：false,AMQP,队列,futao,RabbitMQ,消息,import,com,channel
来源： https://blog.csdn.net/futao__/article/details/91283596

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9

AMQP-RabbitMQ/2/工作队列

2. 工作队列 Work queues

# 个人理解

# 实现工作队列下的非公平队列