RocketMQ简单使用及API解释

HelloWord示例

名词

topic

topic中文意思话题，也就是代表该消息跟什么话题有关，相当于这种消息是什么类型的。

topic是生产者传递消息而消费者提取消息的类别。主题与生产者和消费者之间的关系非常松散。具体来说，一个主题可能有零个，一个或多个向其发送消息的生产者。相反，生产者可以发送不同主题的消息。从消费者的角度来看，一个主题可以由零个，一个或多个消费者组订阅。与此类似，消费者组可以订阅一个或多个主题，只要该组的实例保持其订阅一致即可。

group

能够使用同一个ConsumerGroup名字的消费者必须具有完全相同的订阅才能正确实现负载平衡。它是必需的

ProducerGroup在概念上聚合完全相同规则的所有Producer实例，这在涉及事务消息时尤为重要。

tags

Message构造器里面的可能需要传的参数

他代表了一个sub-topic, 为用户提供了额外的灵活性。使用tag，来自同一业务模块的不同的消息可能具有相同的主题和不同的标签。标签还可以简化RocketMQ提供的查询系统。

就是做了第二层分类

keys

Message构造器里面的可能需要传的参数

主要作用是用来查找消息

maven项目

pom.xml

<dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-client</artifactId>
        <version>4.6.1</version>
 </dependency>

code

//生产消息


    public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {

        //初始化一个消息生产者
        DefaultMQProducer defaultMQProducer = new DefaultMQProducer("group1");
        //配置namesrv地址
        defaultMQProducer.setNamesrvAddr("192.168.47.100:9876");
        //启动
        defaultMQProducer.start();
        //生产消息
        //topic 简单理解该消息是什么类型的，其实对应着多个broker,这跟它的架构有关,每一个broker再注册到namesrv时候会上报自己有哪些topic
        //反过来其实topic就代表了一组broker的地址，如果namesrv没有这个topic的记录，那么会选一台创建topic，namesrv在这里的作用非常大
        // body 就是消息内容
        Message message=new Message("test","hello word".getBytes());
        //发送消息
        defaultMQProducer.send(message);
		defaultMQProducer.shutdown();

    }

//此时登录web控制台查看消息
//选择你的主题，过滤消息，刷新发现刷不出来
//你把时间间隔放大，就出来了，发现StoreTime不对劲，跟虚拟机系统时间有关系
//Centos系统更新时间
/*
yum install ntpdate
ntpdate ntp1.aliyun.com
*/


//消费者
public static void main(String[] args) throws MQClientException {
        //初始化一个消费者
        // 去看DefaultMQPushConsumer实现的接口，下载源码，看方法解释
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer("group2");
        //配置namesrv地址
        defaultMQPushConsumer.setNamesrvAddr("192.168.47.100:9876");

        //每一个consumer只能消费一个topic
        //topic就指定了他去消费一些broker里卖的消息
        //messageSeletor 表示对消息过滤
        //messageSeletor有type和expression，type表示表达式的类型，例如sql92；expression则表示表达式的内容
        //当然也可以直接传表达式,只不过默认表达式类型为tag ，只接收指定tag的消息，当"","*",null,则此topic下的所有消息全部消费
        defaultMQPushConsumer.subscribe("test","*");


        //注册一个消息的监听者,异步消费

        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {

         
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

                msgs.forEach(System.out::println);
                //默认情况下，一个消息只能被组内的一个consumer消费，所以消费完了需要返回消费结果
                //如果是CONSUME_SUCCESS，这样broker收到之后就不会给组内其他consumer消费
                //如果是RECONSUME_LATER，这样broker收到之后延时一段时间给组内的某一个consumer消费，按负载均衡算法来
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        //启动
        defaultMQPushConsumer.start();

        System.out.println("consumer start...");



    }

生产者的api

DefaultMQProducer.class

//send系列    
    
//在同步模式下发送消息。该方法返回只有当发送过程完全完成。
//该方法会有重试机制，一次可能发不成功，当重试次数达到retryTimesWhenSendFailed的大小，就会申明失败
//重点是由于该方法有重试机制，可能会造成多条重复的消息交给broker，程序员自己解决此问题
//最简单的把retryTimesWhenSendFailed的值设置成0就行了
public SendResult send(Message msg) 
//和send(Message msg)方法一样，只不过多了超时时间
public SendResult send(Message msg,long timeout)
//和send(Message msg)方法一样，只不过指定发送到对应的mq    
public SendResult send(Message msg, MessageQueue mq)  
//意思自己猜吧，看参数猜
public SendResult send(Message msg, MessageQueue mq, long timeout)
//和send(Message msg)方法一样，指定对应的MessageQueueSelector，就是选择几个mq发送到其中的一个amq
//arg是配合 MessageQueueSelector一起工作的
public SendResult send(Message msg, MessageQueueSelector selector, Object arg)    
//意思自己猜吧，看参数猜    
public SendResult send(Message msg, MessageQueueSelector selector, Object arg, long timeout)    
    
    
    
//异步发送消息给代理。 
//会立即返回，一旦发送完成，会回调sendCallback
//重点 虽然异步了，但是真正执行发送的线程同样会带有重试机制，程序员自己解决此潜在问题
public void send(Message msg,SendCallback sendCallback)
// 和send(Message msg,SendCallback sendCallback)方法一样，只不过多了超时时间   
public void send(Message msg, SendCallback sendCallback, long timeout)    
//意思自己猜吧，看参数猜   
public void send(Message msg, MessageQueue mq, SendCallback sendCallback)    
//意思自己猜吧，看参数猜      
public void send(Message msg, MessageQueue mq, SendCallback sendCallback, long timeout)   
//意思自己猜吧，看参数猜     
public void send(Message msg, MessageQueueSelector selector, Object arg, SendCallback sendCallback)   
//意思自己猜吧，看参数猜 
public void send(Message msg, MessageQueueSelector selector, Object arg, SendCallback sendCallback, long timeout)    
    
    
//类似于UDP报文传输协议的方式，发完就不管了， 不会等待broker返回的结果
//那么此方法 是不知道这个消息 到底有没有被对方接收的
//好处在于吞吐量大    
 public void sendOneway(Message msg) 
  //意思自己猜吧，看参数猜  
 public void sendOneway(Message msg,MessageQueue mq)   
 //意思自己猜吧，看参数猜   
 public void sendOneway(Message msg, MessageQueueSelector selector, Object arg)
    
//request系列 for rpc   

    
//同步的形式 发送请求，意思就是只有再消费者消费请求给出答复之后才会返回    
//内部实现有重试的机制，可能会消费多个一样的请求，程序员自己解决    
public Message request(final Message msg, final long timeout)  
 //看参数猜，很好猜的       
 public Message request(final Message msg, final MessageQueue mq, final long timeout)    
//看参数猜，很好猜的    
public Message request(final Message msg, final MessageQueueSelector selector, final Object arg,
        final long timeout)     
    
    
//异步请求
//立即返回，当收到回复的消息，回调requestCallback
//与request(final Message msg, final long timeout)一样 ，内部实现有重试的机制，可能会消费多个一样的请求，程序员自己解决        
public void request(final Message msg, final RequestCallback requestCallback, final long timeout)  
   //看参数猜，很好猜的   
public void request(final Message msg, final MessageQueue mq, final RequestCallback requestCallback, long timeout)
 //看参数猜，很好猜的    
 public void request(final Message msg, final MessageQueueSelector selector, final Object arg,
        final RequestCallback requestCallback, final long timeout)  
    
    
    
    
TransactionMQProducer.class    
//sendMessageInTransaction 系列
    
//这个方法是发送事务性消息。    
// 此方法将在版本5.0.0中删除，推荐使用方法sendMessageInTransaction(final Message msg,final Object arg)。   
public TransactionSendResult sendMessageInTransaction(Message msg, LocalTransactionExecuter tranExecuter,final Object arg)
    
    
//这个方法是用来发送事务性消息。 
TransactionSendResult sendMessageInTransaction(Message msg, Object arg)
    
    
    
//批量 for batch
public SendResult send(Collection<Message> msgs) 
public SendResult send(Collection<Message> msgs,long timeout)     
public SendResult send(Collection<Message> msgs,MessageQueue messageQueue)     
public SendResult send(Collection<Message> msgs, MessageQueue messageQueue,long timeout)
官网原话：    
为什么要批处理？
批量发送消息可提高传递小消息的性能    
使用限制
同一批次的消息应具有：相同的主题，相同的waitStoreMsgOK，并且不支持计划。 
此外，一批的总大小不得超过1MB。
    
一般我们不知道自己批量处理的数据大小

生产者的一些其他的api

//这个有空了我会补全

//设置同步模式下 重试次数 
public void setRetryTimesWhenSendFailed(int retryTimesWhenSendFailed)
//设置异步模式下 重试次数 
public void setRetryTimesWhenSendAsyncFailed(final int retryTimesWhenSendAsyncFailed)

批处理

官网原话：    
为什么要批处理？
批量发送消息可提高传递小消息的性能    
使用限制
同一批次的消息应具有：相同的主题，相同的waitStoreMsgOK，并且不支持计划。 
此外，一批的总大小不得超过1MB。
    
一般我们不知道自己批量处理的数据大小   
官方提供的解决方案

public class ListSplitter implements Iterator<List<Message>> {
    private final int SIZE_LIMIT = 1000 * 1000;
    private final List<Message> messages;
    private int currIndex;
    public ListSplitter(List<Message> messages) {
            this.messages = messages;
    }
    @Override public boolean hasNext() {
        return currIndex < messages.size();
    }
    @Override public List<Message> next() {
        int nextIndex = currIndex;
        int totalSize = 0;
        for (; nextIndex < messages.size(); nextIndex++) {
            Message message = messages.get(nextIndex);
            int tmpSize = message.getTopic().length() + message.getBody().length;
            Map<String, String> properties = message.getProperties();
            for (Map.Entry<String, String> entry : properties.entrySet()) {
                tmpSize += entry.getKey().length() + entry.getValue().length();
            }
            tmpSize = tmpSize + 20; //for log overhead 日志相关
            if (tmpSize > SIZE_LIMIT) {
                //单个消息超出大小限制是意外的
                //这里就把单个作为一个list发出去
                if (nextIndex - currIndex == 0) {
                   nextIndex++;  
                }
                break;
            }
            if (tmpSize + totalSize > SIZE_LIMIT) {
                break;
            } else {
                totalSize += tmpSize;
            }
    
        }
        List<Message> subList = messages.subList(currIndex, nextIndex);
        currIndex = nextIndex;
        return subList;
    }
}
//这样就可以把大列表拆分很多的小列表进行发送
ListSplitter splitter = new ListSplitter(messages);
while (splitter.hasNext()) {
   try {
       List<Message>  listItem = splitter.next();
       producer.send(listItem);
   } catch (Exception e) {
       e.printStackTrace();
       //handle the error
   }
}

消费者API

MQConsumer.class
//如果消费失败，消息将被发送回brokers，并延迟一段时间后进行消费，不推荐使用
 void sendMessageBack(final MessageExt msg, final int delayLevel)
 //如果消费失败，消息指定返回某个broker，并延迟一段时间后进行消费，不推荐使用
 void sendMessageBack(final MessageExt msg, final int delayLevel, final String brokerName)    
     
//根据主题来从消费者缓存获取消息队列  
Set<MessageQueue> fetchSubscribeMessageQueues(final String topic)     

    
    
    
MQPushConsumer.class    
    
//订阅一个指定topic，并按subExpression表达式来刷选消息，默认表达式类型为tag
void subscribe(final String topic, final String subExpression) 

//此方法将在版本5.0.0中被删除，因为filterServer已被删除,推荐subscribe(final String topic, final MessageSelector messageSelector)
void subscribe(final String topic, final String fullClassName,final String filterClassSource) 

//订阅一个指定topic，并按selector的表达式来刷选消息，这种表达式可以自定义类型是什么，不懂看MessageSelector类结构
void subscribe(final String topic, final MessageSelector selector)   
    
//退订指定的  topic
 void unsubscribe(final String topic);    
 //更新消息消费线程核心池大小。
public void updateCorePoolSize(int corePoolSize) 


//暂停消费新消息。    
public void suspend()；
//恢复拉取新消息    
public void resume()    
    
    
//注册新消息监听器，消费逻辑就写在  MessageListener里面,不建议使用
void registerMessageListener(MessageListener messageListener); 
//MessageListenerConcurrently并发消息侦听器对象用于并发接收异步交互的消息
//这个里面的回调方法，调用时如果出现问题，不建议抛出异常，而是返回ConsumeConcurrentlyStatus.RECONSUME_LATER
//底层实现是多线程同时回调
//一个queue对应多个线程，就是消息不按queue来区分
void registerMessageListener(final MessageListenerConcurrently messageListener);
// MessageListenerOrderly对象用于按顺序接收异步传递消息。
//底层使用一个队列对应一个线程实现的回调
//重点：多个queue是对应多个线程的
//这个里面的回调方法，调用时如果出现问题，不建议抛出异常，而是返回ConsumeConcurrentlyStatus.RECONSUME_LATER
void registerMessageListener(final MessageListenerOrderly messageListener);


MQPullConsumer.class 同样可以消费消息，但不是回调的方式，是同步获取的方式，没人用
    
重点：无论是MQPullConsumer，还是MQPushConsumer实现，都是主动去询问broker拉取消息，而不是broker主动推，因为由boker主动推的话，会导致consumer堆积太多，来不及消费，然后又不好发送给其他consumer，所以都采用 主动轮询方式
    只不过这两个类 在形式上演示 一个靠拉 一个靠推
    MQPullConsumer就需要自己写代码去拉
    MQPushConsumer帮你写好了拉取的实现

消费者其他的api

//设置消费模式，一个是BROADCASTING（每一个consumer都可以进行消费），一个CLUSTERING（只能被一个consumer消费）
//集群消息是指集群化部署消费者，就是一个组代表一个集群
//当使用集群消费模式时，MQ 认为任意一条消息只需要被集群内的任意一个消费者处理即可。
//特点
//每条消息只需要被处理一次，broker只会把消息发送给消费集群中的一个消费者
//在消息重投时，不能保证路由到同一台机器上
//消费状态由broker维护

//当使用广播消费模式时，MQ 会将每条消息推送给集群内所有注册过的客户端，保证消息至少被每台机器消费一次。
//特点
//消费进度由consumer维护
//保证每个消费者消费一次消息
//消费失败的消息不会重投
public void setMessageModel(MessageModel messageModel)

关于tags

要注意同个一消费组，都订阅同一个topic，但是过滤tags的方式不同，此时消费者有可能会收不到消息

因为broker推送消息是按组来推送到其中的一个，但是被过滤了，这一个组就都不会再推送了，

重点：对于客户端的过滤，消费模式是集群模式，broker按组推送消息，一个组保证推送一次（并且只被该组的某一个Consumer消费）， 只要该组关注了该消息的topic就会推送

验证

 Consumer tag-a
 Consumer tag-b
public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer("group2");                     
        defaultMQPushConsumer.setNamesrvAddr("192.168.47.100:9876");
        defaultMQPushConsumer.subscribe("test","tag-a");//这里换一下
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

                msgs.forEach(System.out::println);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        defaultMQPushConsumer.start();

        System.out.println("consumer tag-a start...");

    }


Producer
public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {

        DefaultMQProducer defaultMQProducer = new DefaultMQProducer("group1");
        defaultMQProducer.setNamesrvAddr("192.168.47.100:9876");
        defaultMQProducer.start();
    //这里换一下tag发送
        Message message=new Message("test","tag-b","hello word".getBytes());
        defaultMQProducer.send(message);

        defaultMQProducer.shutdown();

    }

得出的结果，有时进入到Comsumer tag-a，有时进入Comsumer tag-b，不确定，因为两个consumer是同一组的
    
但是组不同，tag一样，Comsumer tag-a，Comsumer tag-b都会消费

关于sql

MessageSelector

Producer

    public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {

        DefaultMQProducer defaultMQProducer = new DefaultMQProducer("group1");
        defaultMQProducer.setNamesrvAddr("192.168.47.100:9876");
        defaultMQProducer.start();
        Message message=new Message("test","tag-b","hello word".getBytes());
    
        message.putUserProperty("age","100");//使用sql过滤，需要配置这里
    
        defaultMQProducer.send(message);

        defaultMQProducer.shutdown();

    }


Consumer
 public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer("group1");
        defaultMQPushConsumer.setNamesrvAddr("192.168.47.100:9876");
        MessageSelector selector=MessageSelector.bySql("age>101 and age<200");

        defaultMQPushConsumer.subscribe("test",selector);
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

                msgs.forEach(System.out::println);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        defaultMQPushConsumer.start();
        System.out.println("consumer start...");

    }

此时启动Consumer会报错
Exception in thread "main" org.apache.rocketmq.client.exception.MQClientException: CODE: 1  DESC: The broker does not support consumer to filter message by SQL92
 说是服务器不支持 SQL92过滤消息
  please visit the url, http://rocketmq.apache.org/docs/faq/ 点进去简单找是没有找到解决方案的

进入虚拟机，

修改broker配置文件

cd /opt/wyz/rocketmq-4.7.1-SNAPSHOT #rocket的安装路径
cd conf/
vi broker.conf 
######
#再最后面加上
enablePropertyFilter=true

######

cd ../bin/
#加载一下配置文件
./mqbroker -n localhost:9876 -c ../conf/broker.conf


#此时去跑Comsumer就可以成功

关于事务消息

分布式系统中的事务可以使用TCC（Try、Confirm、Cancel）、2pc来解决分布式系统中的消息原子性

RocketMQ 4.3+提供分布事务功能，通过 RocketMQ 事务消息能达到分布式事务的最终一致

RocketMQ实现方式

Half Message：预处理消息，当broker收到此类消息后，会存储到RMQ_SYS_TRANS_HALF_TOPIC的消息消费队列中

检查事务状态：Broker会开启一个定时任务，消费RMQ_SYS_TRANS_HALF_TOPIC队列中的消息，每次执行任务会向消息发送者确认事务执行状态（提交、回滚、未知），如果是未知，等待下一次回调。

超时：如果超过回查次数，默认回滚消息

TransactionListener

//半事务消息还是一条消息，只不过该消息还不能被消费，由broker维护状态
//当发送半事务消息成功时，将调用此方法来执行本地事务。
LocalTransactionState executeLocalTransaction(final Message msg, final Object arg);

// 半事务消息发送给broker成功回调执行本地事务时 ，在一定时间内得不到的响应。broker将发送检查消息来检查事务状态，并调用此方法来获取本地事务状态
    LocalTransactionState checkLocalTransaction(final MessageExt msg);

测试

public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {


        TransactionMQProducer producer = new TransactionMQProducer("group1");
        producer.setNamesrvAddr("192.168.47.100:9876");
        producer.start();
        Message message=new Message("demo","hello word".getBytes());
        message.putUserProperty("age","150");



        //设置事务监听
        producer.setTransactionListener(new TransactionListener() {
            /**
             *
             * @param msg 就是发送的那条半事务消息
             * @param arg 回调时执行本地事务的资源
             * @return
             */
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {

                try {
                    System.out.println("arg:"+arg);
                    System.out.println("MsgBody:"+new String(msg.getBody(),"utf-8"));
                    System.out.println("TransactionId:"+msg.getTransactionId());
                    Thread.sleep(20000);
//                    int i=1/0;
                    System.out.println("本地事务执行完成，需要提交");
                    return LocalTransactionState.COMMIT_MESSAGE;
                }catch (Exception ex){
                    System.out.println("出错了，需要回滚");
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }

            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                return LocalTransactionState.UNKNOW;
            }
        });

        //发送事务消息
        //arg 就是等下半事务消息发送成功执行回调 进行本地事务所需要的资源
        //message就是半事务消息，这是由broker来把他变成半事务消息的
        producer.sendMessageInTransaction(message,"我是本地事务需要的资源");

        producer.shutdown();

    }
先不启动消费者，发现整个流程都可以走完

说明producer 事务的执行与 consumer没有任何关系，producer默认了半事务消息进入broker之后
回调执行本地事务，执行成功返回COMMIT_MESSAGE状态，broker把半事务消息变成可消费消息，producer会默认该消息一定被执行，最终一致性
    
由这个过程，说明producer与broker 之间的分布式事务 是由 两阶段提交，（具体是不是，得看源码）
    第一阶段  producer 发送半事务消息， broker 收到消息 维护消息状态并且 刷到磁盘（不刷到磁盘无法保证一致性），刷盘成功后，做ack（应答），producer会有一个T线程 专门盯着这个ack有没有来 ，一旦来了，该T线程就会回调监听器得方法，执行本地事务  
    
    第二阶段 要么返回commit状态，要么返回rollback状态，T线程会把状态发给broker，而broker根据状态进行处理，要么变成可消费消息，要么放入回收站
    这个T线程其实就是调用sendMessageInTransaction方法的线程，重点：发送到回调，这个过程是同步的
    
    在两个阶段中有一个中间状态，就是在执行本地事务时，（后面文字都是猜的，讲道理应该是这样）卡住了，或者死机了，那么broker会开一个定时任务，来访问producer。1：producer挂了，连不上，那么这个定时任务就多跑几次，超过一定次数，这个半事务消息进入回收站，2：卡住了，回调健康检查函数，这里可以返回状态，但是一般返回unknow状态，那么这个定时任务就多跑几次，超过一定次数，这个半事务消息进入回收站。
    调用检查回调this.checkExecutor.submit(request);
    这个checkExecutor通过public void setExecutorService(ExecutorService executorService)方法设置
    
Consumer什么时候启动，什么时候消费其实无所谓，反正消息刷盘了，对应得Consumer一上线就能消费到

重试机制

producer

默认超时时间

   /**
    * Timeout for sending messages.
    */
   private int sendMsgTimeout = 3000;
// 异步发送时 重试次数，默认 2
producer.setRetryTimesWhenSendAsyncFailed(1);
// 同步发送时 重试次数，默认 2
producer.setRetryTimesWhenSendFailed(1);	

// 是否向其他broker发送请求 默认false
producer.setRetryAnotherBrokerWhenNotStoreOK(true);

Consumer

消费超时，单位分钟

//如果不是并发消费模式，一条消息的消费过长，会影响后续的消费，所以设置超时事件
consumer.setConsumeTimeout()

发送ack，消费失败

RECONSUME_LATER

broker投递

只有在消息模式为MessageModel.CLUSTERING集群模式时，Broker才会自动进行重试，广播消息不重试

重投使用messageDelayLevel

默认值

messageDelayLevel	1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h

总结：

我们使用rocketMQ，那我们就是客户，我们需要知道

怎么投消息，producer 一些列api，投消息很简单

broker怎么推送消息： 这个至关重要，不懂这个，编码会产生重复消费

MessageModel.CLUSTERING集群模式：

broker的推送消息，加入推送消息A，消息A的topic是a

（猜的）大概流程 去访问namesrv，把关注了a这个topic的consumer全部拿过来，所有的consumer按group分开，遍历每一个组 ，每个组用负载均衡的算法选择该组的一个consumer 进行推送

消息模式是广播就不用多说了

赏

谢谢你请我吃糖果

支付宝

微信

• 本文作者： 忘忧症
• 本文链接： https://NepenthesZGW.github.io/2020/06/17/framework/RocketMQ/RocketMQ简单使用/
• 版权声明： 本博客所有文章除特别声明外，均采用 MIT 许可协议。转载请注明出处！