文章
文章用户文档快讯圈子网址导航

AMQP-RabbitMQ/3/发布订阅模式

释放双眼，带上耳机，听听看~！

3. 发布订阅模式 Publish/Subscribe – 全集监听fanout

一次向多个消费者发送消息

  • 图示

个人理解

  • 生产者定义Exchange，同时将Exchange的类型定义为fanout，并向该Exchange发送消息。
  • 消费者定义队列Queue，并将队列与该交换机进行绑定。之后交换机付负责将消息全量推送给每一个与之绑定的Queue

**RabbitMQ中消息传递模型的核心思想是生产者永远不会将任何消息直接发送到队列。实际上，生产者通常甚至不知道消息是否会被传递到任何队列。
相反，生产者只能向Exchange发送消息。Exchange所做的工作非常简单。一方面，它接收来自生产者的消息，另一方面将它们推送到队列。Exchange必须确切知道如何处理它收到的消息。它应该附加到特定队列吗？它应该附加到多个队列吗？或者它应该被丢弃。其规则由交换类型定义 。**
有几种交换类型可供选择：direct ,topic ,headers andfanout

  • fanout: 将它接收到的消息广播到所有绑定到它的消息队列上。(忽略路由键routingKey)

  • 生产者 – 发布者



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
1package com.futao.springmvcdemo.mq.rabbit.ps;

2

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

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

5import com.rabbitmq.client.BuiltinExchangeType;

6import com.rabbitmq.client.Channel;

7import com.rabbitmq.client.Connection;

8import lombok.Cleanup;

9import lombok.SneakyThrows;

10import lombok.extern.slf4j.Slf4j;

11

12/**

13 * 发布订阅-发布者

14 *

15 * @author futao

16 * Created on 2019-04-22.

17 */

18@Slf4j

19public class Publisher {

20    @SneakyThrows

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

22        @Cleanup

23        Connection connection = RabbitMqConnectionTools.getConnection();

24        @Cleanup

25        Channel channel = connection.createChannel();

26        //定义交换器类型

27        channel.exchangeDeclare(ExchangeTypeEnum.FANOUT.getExchangeName(), BuiltinExchangeType.FANOUT);

28        String msg = "Hello RabbitMq!";

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

30            channel.basicPublish(ExchangeTypeEnum.FANOUT.getExchangeName(), "", null, (msg + i).getBytes());

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

32        }

33    }

34}

35

  • 消费者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
42
43
44
45
46
47
48
49
1package com.futao.springmvcdemo.mq.rabbit.ps;

2

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

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

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

6import com.rabbitmq.client.BuiltinExchangeType;

7import com.rabbitmq.client.Channel;

8import com.rabbitmq.client.DeliverCallback;

9import lombok.SneakyThrows;

10import lombok.extern.slf4j.Slf4j;

11

12/**

13 * 发布订阅-订阅者

14 *

15 * @author futao

16 * Created on 2019-04-22.

17 */

18@Slf4j

19public class SubscriberOne {

20    @SneakyThrows

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

22        Channel channel = RabbitMqConnectionTools.getChannel();

23        //开启持久化队列

24        boolean durable = true;

25        channel.queueDeclare(RabbitMqQueueEnum.EXCHANGE_QUEUE_FANOUT_ONE.getQueueName(), durable, false, false, null);

26        //定义交换器类型

27        channel.exchangeDeclare(ExchangeTypeEnum.FANOUT.getExchangeName(), BuiltinExchangeType.FANOUT);

28        //将消息队列与Exchange交换器绑定

29        channel.queueBind(RabbitMqQueueEnum.EXCHANGE_QUEUE_FANOUT_ONE.getQueueName(), ExchangeTypeEnum.FANOUT.getExchangeName(), "");

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

31        channel.basicQos(1);

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

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

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

35            //acknowledgment应答

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

37            try {

38                Thread.sleep(1000);

39            } catch (Exception e) {

40

41            }

42        });

43        //关闭自动应答

44        boolean autoAck = false;

45        channel.basicConsume(RabbitMqQueueEnum.EXCHANGE_QUEUE_FANOUT_ONE.getQueueName(), autoAck, deliverCallback, consumerTag -> {

46        });

47    }

48}

49

  • 消费者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
42
43
44
45
46
47
48
49
1package com.futao.springmvcdemo.mq.rabbit.ps;

2

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

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

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

6import com.rabbitmq.client.BuiltinExchangeType;

7import com.rabbitmq.client.Channel;

8import com.rabbitmq.client.DeliverCallback;

9import lombok.SneakyThrows;

10import lombok.extern.slf4j.Slf4j;

11

12/**

13 * 发布订阅-订阅者

14 *

15 * @author futao

16 * Created on 2019-04-22.

17 */

18@Slf4j

19public class SubscriberTwo {

20    @SneakyThrows

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

22        Channel channel = RabbitMqConnectionTools.getChannel();

23        //开启持久化队列

24        boolean durable = true;

25        channel.queueDeclare(RabbitMqQueueEnum.EXCHANGE_QUEUE_FANOUT_TWO.getQueueName(), durable, false, false, null);

26        //定义交换器类型为fanout

27        channel.exchangeDeclare(ExchangeTypeEnum.FANOUT.getExchangeName(), BuiltinExchangeType.FANOUT);

28        //将消息队列与Exchange交换器进行绑定

29        channel.queueBind(RabbitMqQueueEnum.EXCHANGE_QUEUE_FANOUT_TWO.getQueueName(), ExchangeTypeEnum.FANOUT.getExchangeName(), "");

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

31        channel.basicQos(1);

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

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

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

35            //acknowledgment应答

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

37            try {

38                Thread.sleep(2000);

39            } catch (Exception e) {

40

41            }

42        });

43        //关闭自动应答

44        boolean autoAck = false;

45        channel.basicConsume(RabbitMqQueueEnum.EXCHANGE_QUEUE_FANOUT_TWO.getQueueName(), autoAck, deliverCallback, consumerTag -> {

46        });

47    }

48}

49

  • 注意

  • 没有在发布者定义队列，而是定义了交换器Exchange。发布者将消息发送到Exchange，而不是Queue

    • 在订阅者端，每个订阅者定义了自己的消息队列，并且将自己的消息队列与Exchange进行绑定。则在每次发布者向相应的Exchange发送消息的时候，Exchange会将消息发送至订阅了该Exchange的队列。(即：每个订阅者收到的消息都是一样的)

  • 测试结果



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
1>>> 订阅者1

2[main] INFO mq.rabbit.ps.SubscriberOne - Waiting for message...

3[pool-1-thread-4] INFO mq.rabbit.ps.SubscriberOne - 收到消息:[Hello RabbitMq!0],tag:[amq.ctag-Fc_B_CoCYUBoBhEcOlC7vw]

4[pool-1-thread-5] INFO mq.rabbit.ps.SubscriberOne - 收到消息:[Hello RabbitMq!1],tag:[amq.ctag-Fc_B_CoCYUBoBhEcOlC7vw]

5[pool-1-thread-6] INFO mq.rabbit.ps.SubscriberOne - 收到消息:[Hello RabbitMq!2],tag:[amq.ctag-Fc_B_CoCYUBoBhEcOlC7vw]

6[pool-1-thread-7] INFO mq.rabbit.ps.SubscriberOne - 收到消息:[Hello RabbitMq!3],tag:[amq.ctag-Fc_B_CoCYUBoBhEcOlC7vw]

7[pool-1-thread-8] INFO mq.rabbit.ps.SubscriberOne - 收到消息:[Hello RabbitMq!4],tag:[amq.ctag-Fc_B_CoCYUBoBhEcOlC7vw]

8[pool-1-thread-9] INFO mq.rabbit.ps.SubscriberOne - 收到消息:[Hello RabbitMq!5],tag:[amq.ctag-Fc_B_CoCYUBoBhEcOlC7vw]

9[pool-1-thread-10] INFO mq.rabbit.ps.SubscriberOne - 收到消息:[Hello RabbitMq!17],tag:[amq.ctag-Fc_B_CoCYUBoBhEcOlC7vw]

10[pool-1-thread-22] INFO mq.rabbit.ps.SubscriberOne - 收到消息:[Hello RabbitMq!18],tag:[amq.ctag-Fc_B_CoCYUBoBhEcOlC7vw]

11[pool-1-thread-23] INFO mq.rabbit.ps.SubscriberOne - 收到消息:[Hello RabbitMq!19],tag:[amq.ctag-Fc_B_CoCYUBoBhEcOlC7vw]

12

13

14>>> 订阅者2

15[main] INFO mq.rabbit.ps.SubscriberTwo - Waiting for message...

16[pool-1-thread-4] INFO mq.rabbit.ps.SubscriberTwo - 收到消息:[Hello RabbitMq!0],tag:[amq.ctag-ip59jtcKJBQFC2KU9DperQ]

17[pool-1-thread-5] INFO mq.rabbit.ps.SubscriberTwo - 收到消息:[Hello RabbitMq!1],tag:[amq.ctag-ip59jtcKJBQFC2KU9DperQ]

18[pool-1-thread-6] INFO mq.rabbit.ps.SubscriberTwo - 收到消息:[Hello RabbitMq!2],tag:[amq.ctag-ip59jtcKJBQFC2KU9DperQ]

19[pool-1-thread-7] INFO mq.rabbit.ps.SubscriberTwo - 收到消息:[Hello RabbitMq!3],tag:[amq.ctag-ip59jtcKJBQFC2KU9DperQ]

20[pool-1-thread-8] INFO mq.rabbit.ps.SubscriberTwo - 收到消息:[Hello RabbitMq!4],tag:[amq.ctag-ip59jtcKJBQFC2KU9DperQ]

21[pool-1-thread-9] INFO mq.rabbit.ps.SubscriberTwo - 收到消息:[Hello RabbitMq!5],tag:[amq.ctag-ip59jtcKJBQFC2KU9DperQ]

22[pool-1-thread-10] INFO mq.rabbit.ps.SubscriberTwo - 收到消息:[Hello RabbitMq!17],tag:[amq.ctag-ip59jtcKJBQFC2KU9DperQ]

23[pool-1-thread-22] INFO mq.rabbit.ps.SubscriberTwo - 收到消息:[Hello RabbitMq!18],tag:[amq.ctag-ip59jtcKJBQFC2KU9DperQ]

24[pool-1-thread-23] INFO mq.rabbit.ps.SubscriberTwo - 收到消息:[Hello RabbitMq!19],tag:[amq.ctag-ip59jtcKJBQFC2KU9DperQ]

25

Related posts:

  1. ASP.NET Core微服务之基于Steeltoe使用Zipkin实现分布式追踪
  2. Springboot+Dubbo+Nacos 注解方式实现微服务调用
  3. Netty游戏服务器实战开发(1):Netty 的自带解码器
  4. Netty游戏服务器实战开发(10):Netty结合kafka实现分布式消息队列

给TA打赏
共{{data.count}}人
人已打赏
[db:标签]
安全网络

CDN安全市场到2022年价值76.3亿美元

2018-2-1 18:02:50

安全网络

分布式锁

2021-8-18 16:36:11

猜你喜欢
❯

解锁会员权限

开通会员

解锁海量优质VIP资源

立刻开通
个人中心
购物车
优惠劵
今日签到
有新私信 私信列表
搜索

  • 扫码打开当前页

返回顶部