AMQP-RabbitMQ/2/工作队列

2. 工作队列 Work queues

Distributing tasks among workers

消息将发送给c1或者c2

个人理解

• 生产者定义Queue，并向该队列发送消息

• 多个消费者可以从指定的同一个Queue中读取消息。每条消息只会发送给其中某一个消费者。

• 生产者

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
1package com.futao.springmvcdemo.mq.rabbit.workqueue;

2

3import com.futao.springmvcdemo.mq.rabbit.RabbitMqConnectionTools;

4import com.futao.springmvcdemo.mq.rabbit.RabbitMqQueueEnum;

5import com.rabbitmq.client.Channel;

6import com.rabbitmq.client.Connection;

7import lombok.Cleanup;

8import lombok.SneakyThrows;

9import lombok.extern.slf4j.Slf4j;

10

11/**

12 * 简单发送者

13 *

14 * @author futao

15 * Created on 2019-04-22.

16 */

17@Slf4j

18public class Send {

19    @SneakyThrows

20    public static void main(String[] args) {

21        @Cleanup

22        Connection connection = RabbitMqConnectionTools.getConnection();

23        @Cleanup

24        Channel channel = connection.createChannel();

25        //开启持久化消息

26        boolean durable = true;

27        //定义一个队列

28        channel.queueDeclare(RabbitMqQueueEnum.WORK_QUEUE.getQueueName(), durable, false, false, null);

29        String msg = "Hello RabbitMq!";

30        for (int i = 0; i < 20; i++) {

31            channel.basicPublish("", RabbitMqQueueEnum.WORK_QUEUE.getQueueName(), null, (msg + i).getBytes());

32            log.info("Send msg:[{}] success", (msg + i));

33        }

34    }

35}

36

• 消费者1 – 每1秒处理一条

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
1package com.futao.springmvcdemo.mq.rabbit.workqueue;

2

3import com.futao.springmvcdemo.mq.rabbit.RabbitMqConnectionTools;

4import com.futao.springmvcdemo.mq.rabbit.RabbitMqQueueEnum;

5import com.rabbitmq.client.Channel;

6import com.rabbitmq.client.DeliverCallback;

7import lombok.SneakyThrows;

8import lombok.extern.slf4j.Slf4j;

9

10/**

11 * 简单消费者

12 *

13 * @author futao

14 * Created on 2019-04-22.

15 */

16@Slf4j

17public class RecvOne {

18    @SneakyThrows

19    public static void main(String[] args) {

20        Channel channel = RabbitMqConnectionTools.getChannel();

21        //开启持久化消息

22        boolean durable = true;

23        channel.queueDeclare(RabbitMqQueueEnum.WORK_QUEUE.getQueueName(), durable, false, false, null);

24        log.info("Waiting for message...");

25        DeliverCallback deliverCallback = ((consumerTag, message) -> {

26            log.info("收到消息:[{}],tag:[{}]", new String(message.getBody()), consumerTag);

27            //acknowledgment应答

28            channel.basicAck(message.getEnvelope().getDeliveryTag(), false);

29            try {

30                Thread.sleep(1000);

31            } catch (Exception e) {

32

33            }

34        });

35        //关闭自动应答

36        boolean autoAck = false;

37        channel.basicConsume(RabbitMqQueueEnum.WORK_QUEUE.getQueueName(), autoAck, deliverCallback, consumerTag -> {

38        });

39    }

40}

41

• 消费者2 – 每2秒处理一条

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
1package com.futao.springmvcdemo.mq.rabbit.workqueue;

2

3import com.futao.springmvcdemo.mq.rabbit.RabbitMqConnectionTools;

4import com.futao.springmvcdemo.mq.rabbit.RabbitMqQueueEnum;

5import com.rabbitmq.client.Channel;

6import com.rabbitmq.client.DeliverCallback;

7import lombok.SneakyThrows;

8import lombok.extern.slf4j.Slf4j;

9

10/**

11 * 简单消费者

12 *

13 * @author futao

14 * Created on 2019-04-22.

15 */

16@Slf4j

17public class RecvTwo {

18    @SneakyThrows

19    public static void main(String[] args) {

20        Channel channel = RabbitMqConnectionTools.getChannel();

21        //开启持久化消息

22        boolean durable = true;

23        channel.queueDeclare(RabbitMqQueueEnum.WORK_QUEUE.getQueueName(), durable, false, false, null);

24        log.info("Waiting for message...");

25        DeliverCallback deliverCallback = ((consumerTag, message) -> {

26            log.info("收到消息:[{}],tag:[{}]", new String(message.getBody()), consumerTag);

27            //acknowledgment应答

28            channel.basicAck(message.getEnvelope().getDeliveryTag(), false);

29            try {

30                Thread.sleep(2000);

31            } catch (Exception e) {

32

33            }

34        });

35        //关闭自动应答

36        boolean autoAck = false;

37        channel.basicConsume(RabbitMqQueueEnum.WORK_QUEUE.getQueueName(), autoAck, deliverCallback, consumerTag -> {

38        });

39    }

40}

41

• 结果

• 生产者

  • 消费者1日志

  • 消费者2日志

• 特点: 多个消费者之间，不论消息的处理速度，都是平均分发(公平分发)。你一个，我一个，他一个。此时是公平队列

• 注意：

• 定义队列的时候，设置是否开启消息的持久化（该设置需要同时在生产者和消费者设置）

1
2
3
4
1       //开启持久化消息

2        boolean durable = true;

3        channel.queueDeclare(RabbitMqQueueEnum.WORK_QUEUE.getQueueName(), durable, false, false, null);

4

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

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

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

1
2
3
1        //告诉rabbitmq一次只发送一条消息，并且在前一个消息未被处理或者消费之前，不继续发送下一个消息

2        channel.basicQos(1);

3

• 测试结果

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