RocketMQ（05）——消息的群集消费和广播消费

悠悠 2023-07-03 04:39 30阅读 0赞

# 消息的群集消费和广播消费 #

RocketMQ的消费者进行消息消费时有两种消费方式，群集消费和广播消费。默认是群集消费。

## 群集消费 ##

Consumer都有一个Group，当两个Consumer实例属于同一个Group时，它们会共享消息队列中的消息消费位移，即同一条消息只会由一个消费者实例消费。实际上一个队列只会分配给一个消费者实例，那么属于该队列中的消息就只能被一个消费者实例消费了。当一个消费者实例挂了后，会重新为消费者实例分配队列，这样原本分配给挂了的那个实例的队列中又会分配给其它消费者进行消费。RocketMQ限制了同一个JVM中不允许有相同Group名称的Consumer实例存在，所以同一Group的多个Consumer往往是部署在不同机器上的，通常是同一程序部署了多份。假设现在有名为group1的Consumer在Machine1上部署了一份，在Machine2上部署了一份，它们订阅的消息队列中有消息1-10共10条消息，那么可能1-5条消息由Machine1上的消费者消费，6-10条消息由Machine2上的消费者消费。如果在Machine1上还部署了一个名为group2的消费者，则该消费者可以从消息队列中消费1-10条消息，因为它跟group1是不同的Group，不同Group的消费者对同一队列的消费是完全独立的。比如下面的代码对应的就是群集消费。

@Test
    public void testConsumer() throws Exception { 
      DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_group1");
      consumer.setNamesrvAddr(nameServer);
      consumer.subscribe("topic1", "tag1");
      consumer.registerMessageListener(new MessageListenerConcurrently() { 
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { 
          System.out.println(Thread.currentThread().getName() + "收到了消息" + msgs);
          return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
      });
      consumer.start();
      TimeUnit.SECONDS.sleep(120);
    }

## 广播消费 ##

消费者的消费模式是广播消费时，每一个消费者实例对消息队列中消息的消费是独立的，而不管它们的Group是什么。假设现在有名为group1的Consumer在Machine1上部署了一份，在Machine2上部署了一份，它们订阅的消息队列中有消息1-10共10条消息，那么可能1-10条消息由Machine1上的消费者消费，它们也会由Machine2上的消费者消费。可以通过消费者的`setMessageModel(MessageModel.BROADCASTING)`指定消息的消费模式为广播消费，它的默认值是CLUSTERING。

@Test
    public void testBroadcastConsume() throws Exception { 
      String topic = "topic1";
      String tag = "tag1";
      String consumerGroup = "consumer-group1";
      DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
      consumer.setNamesrvAddr(this.nameServer);
      //广播方式，同一消息可以被所有的消费者消费。
      consumer.setMessageModel(MessageModel.BROADCASTING);
      try { 
        consumer.subscribe(topic, tag);
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> { 
          System.out.println("消费了消息——" + msgs.get(0).getMsgId());
          return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
      } catch (MQClientException e) { 
        e.printStackTrace();
      }
      TimeUnit.SECONDS.sleep(Integer.MAX_VALUE);
    }

（注：本文是基于RocketMQ4.5.0所写）

发表评论取消回复

表情：

评论列表（有 0 条评论，30人围观）

还没有评论，来说两句吧...

相关阅读

相关 RocketMQ（05）——消息的群集消费和广播消费

消息的群集消费和广播消费 RocketMQ的消费者进行消息消费时有两种消费方式，群集消费和广播消费。默认是群集消费。群集消费 Consumer都有一个Group

悠悠/ 2023年07月03日 04:39/ 0 赞/ 31 阅读

相关消息队列：集群消费，广播消费，Topic、Broker。

Topic 主题，从逻辑上讲一个Topic就是一个Queue，即一个队列；从存储上讲，一个Topic存储了一类相同的消息，是一类消息的集合。比如一个名称为trade.or

亦凉/ 2023年06月21日 11:25/ 0 赞/ 101 阅读

相关 RocketMQ消息消费（三）

顺序消息 RocketMQ支持局部消息顺序消费，可以确保同一个消息消费队列中的消息被顺序消费，如果需要做到全局顺序消费则可以将主题配置成一个队列，例如数据库BinLog等

我就是我/ 2023年01月19日 08:09/ 0 赞/ 385 阅读

相关 RocketMQ消息消费（二）

消息队列负载与重新分布机制 RocketMQ消息队列重新分布是由RebalanceService线程来实现的。一个MQClient-Instance持有一个Rebalan

拼搏现实的明天。/ 2023年01月19日 04:36/ 0 赞/ 372 阅读

相关 RocketMQ消息消费（一）

消息消费以组的模式开展，一个消费组内可以包含多个消费者，每一个消费组可订阅多个主题，消费组之间有集群模式与广播模式两种消费模式。集群模式，主题下的同一条消息只允许被其中一个消费

迈不过友情╰/ 2023年01月17日 14:54/ 0 赞/ 530 阅读

相关 Celery: 实现集群消费和广播消费

集群消费集群消费类似于负载均衡，生产产生的消息只会被一个worker消费一次 ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGV

雨点打透心脏的1/2处/ 2022年12月03日 13:08/ 0 赞/ 234 阅读

相关 RocketMq 消息发送和消息消费机制

消费发送生产者向消息队列里写入消息，不同的业务场景需要生产者采用不同的写入策略。 RocketMQ提供了吴中发送策略比如同步发送、异步发送、Oneway发送、延迟发送

桃扇骨/ 2022年09月04日 12:43/ 0 赞/ 501 阅读

相关 RocketMQ的顺序消费和事务消费

一、三种消费：1.普通消费 2. 顺序消费 3.事务消费 1.1 顺序消费：在网购的时候，我们需要下单，那么下单需要假如有三个顺序，第一、创建订单，第二：订单付款，第三

╰+哭是因爲堅強的太久メ/ 2022年05月09日 04:22/ 0 赞/ 332 阅读

相关 RocketMQ——广播消费模式与集群消费模式

RocketMQ有两种消费模式：BROADCASTING广播模式，CLUSTERING集群模式，默认的是集群消费模式。本博客主要以广播模式为例！！！ 1.简介广播消

你的名字/ 2021年09月27日 00:04/ 0 赞/ 681 阅读

相关 RocketMQ广播消费与集群消费

1.集群消费方式一个ConsumerGroup中的Consumer实例平均分摊消费生产者发送的消息。例如某个Topic有九条消息，其中一个Consumer Group有三

左手的ㄟ右手/ 2021年06月10日 20:40/ 0 赞/ 757 阅读