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

你的名字 2021-09-27 00:04 621阅读 0赞

RocketMQ有两种消费模式：BROADCASTING广播模式，CLUSTERING集群模式，默认的是 集群消费模式。

本博客主要以广播模式为例！！！

1.简介

广播消费指的是：一条消息被多个consumer消费，即使这些consumer属于同一个ConsumerGroup,消息也会被ConsumerGroup中的每个Consumer都消费一次，广播消费中ConsumerGroup概念可以认为在消息划分方面无意义。

在CORBA Notification规范中，消费方式都属于广播消费。

在JMS规范中，相当于JMS publish/subscribe model

集群消费模式：一个ConsumerGroup中的Consumer实例平均分摊消费消息。例如某个Topic有9条消息，其中一个ConsumerGroup有3个实例（可能是3个进程，或者3台机器），那么每个实例只消费其中部分，消费完的消息不能被其他实例消费。

在CORBA Notification规范中，无此消费方式。

在JMS规范中，JMS point-to-point model与之类似，但是RocketMQ的集群消费功能大等于PTP模型。因为RocketMQ单个ConsumerGroup内的消费者类似于PTP，但是一个Topic/Queue可以被多个ConsumerGroup消费。

2.实例

（1）Producer

package com.gwd.rocketmq;
    
    import com.alibaba.rocketmq.client.exception.MQClientException;
    import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
    import com.alibaba.rocketmq.client.producer.SendResult;
    import com.alibaba.rocketmq.common.message.Message;
    
    /** 
    * @FileName Producer.java
    * @Description:
    * @author gu.weidong
    * @version V1.0
    * @createtime 2018年6月25日 上午9:48:37 
    * 修改历史：
    * 时间           作者          版本        描述
    *====================================================  
    *
    */
    public class Producer {
    	public static void main(String[] args) throws MQClientException, InterruptedException {  
            //声明并初始化一个producer  
            //需要一个producer group名字作为构造方法的参数，这里为producer1  
            DefaultMQProducer producer = new DefaultMQProducer("producer1");  
              
            //设置NameServer地址,此处应改为实际NameServer地址，多个地址之间用；分隔  
            //NameServer的地址必须有，但是也可以通过环境变量的方式设置，不一定非得写死在代码里  
            producer.setNamesrvAddr("192.168.140.128:9876;192.168.140.129:9876");  
            //调用start()方法启动一个producer实例  
            producer.start();  
      
            //发送10条消息到Topic为TopicTest，tag为TagA，消息内容为“Hello RocketMQ”拼接上i的值  
            for (int i = 0; i < 100; i++) {  
                try {  
                    Message msg = new Message("TopicTest",// topic  
                            "TagA",// tag  
                            ("Hello RocketMQ " + i).getBytes("utf-8")// body  
                    );  
                      
                    //调用producer的send()方法发送消息  
                    //这里调用的是同步的方式，所以会有返回结果  
                    SendResult sendResult = producer.send(msg);  
                    System.out.println(sendResult.getSendStatus()); //发送结果状态  
                    //打印返回结果，可以看到消息发送的状态以及一些相关信息  
                    System.out.println(sendResult);  
                } catch (Exception e) {  
                    e.printStackTrace();  
                    Thread.sleep(1000);  
                }  
            }  
      
            //发送完消息之后，调用shutdown()方法关闭producer  
            producer.shutdown();  
        }  
    }

（2）Consumer1

package com.gwd.rocketmq;
    
    import java.text.SimpleDateFormat;
    import java.util.Date;
    import java.util.List;
    
    import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
    import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
    import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
    import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
    import com.alibaba.rocketmq.client.exception.MQClientException;
    import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
    import com.alibaba.rocketmq.common.message.MessageExt;
    import com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;
    
    /** 
    * @FileName Consumer.java
    * @Description:
    * @author gu.weidong
    * @version V1.0
    * @createtime 2018年6月25日 上午9:49:39 
    * 修改历史：
    * 时间           作者          版本        描述
    *====================================================  
    *
    */
    public class Consumer {
    	public static void main(String[] args) throws MQClientException {  
            
            //声明并初始化一个consumer  
             //需要一个consumer group名字作为构造方法的参数，这里为consumer1  
             DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer1");  
      
             //同样也要设置NameServer地址  
             consumer.setNamesrvAddr("192.168.140.128:9876;192.168.140.129:9876");  
             consumer.setMessageModel(MessageModel.BROADCASTING); 
             consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);  
      
             //设置consumer所订阅的Topic和Tag，*代表全部的Tag  
             consumer.subscribe("TopicTest", "*");  
      
             //设置一个Listener，主要进行消息的逻辑处理  
             consumer.registerMessageListener(new MessageListenerConcurrently() {  
      
                 @Override  
                 public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,  
                                                                 ConsumeConcurrentlyContext context) {  
                     for (MessageExt messageExt : msgs) {    
                        String messageBody = new String(messageExt.getBody());   
                         System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(
                        		 new Date())+"消费响应：msgId : " + messageExt.getMsgId() + ",  msgBody : " + messageBody);//输出消息内容    
                     }    
                       
                     //返回消费状态  
                     //CONSUME_SUCCESS 消费成功  
                     //RECONSUME_LATER 消费失败，需要稍后重新消费  
                     return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;  
                 }  
             });  
      
             //调用start()方法启动consumer  
             consumer.start();  
             System.out.println("Consumer Started.");  
         }  
    }

(3)consumer2

package com.gwd.rocketmq;
    
    import java.text.SimpleDateFormat;
    import java.util.Date;
    import java.util.List;
    
    import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
    import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
    import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
    import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
    import com.alibaba.rocketmq.client.exception.MQClientException;
    import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
    import com.alibaba.rocketmq.common.message.MessageExt;
    import com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;
    
    /** 
    * @FileName Consumer2.java
    * @Description:
    * @author gu.weidong
    * @version V1.0
    * @createtime 2018年6月25日 上午11:26:23 
    * 修改历史：
    * 时间           作者          版本        描述
    *====================================================  
    *
    */
    public class Consumer2 {
    public static void main(String[] args) throws MQClientException {  
            
            //声明并初始化一个consumer  
             //需要一个consumer group名字作为构造方法的参数，这里为consumer1  
             DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer1");  
      
             //同样也要设置NameServer地址  
             consumer.setNamesrvAddr("192.168.140.128:9876;192.168.140.129:9876");  
             consumer.setMessageModel(MessageModel.BROADCASTING);
             consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);  
             //设置consumer所订阅的Topic和Tag，*代表全部的Tag  
             consumer.subscribe("TopicTest", "*");  
             //设置一个Listener，主要进行消息的逻辑处理  
             consumer.registerMessageListener(new MessageListenerConcurrently() {  
                 @Override  
                 public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,  
                                                                 ConsumeConcurrentlyContext context) {  
                     for (MessageExt messageExt : msgs) {    
                        String messageBody = new String(messageExt.getBody());    
                         System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(
                        		 new Date())+"2--------消费响应：msgId : " + messageExt.getMsgId() + ",  msgBody : " + messageBody);//输出消息内容    
                     }    
                     //返回消费状态  
                     //CONSUME_SUCCESS 消费成功  
                     //RECONSUME_LATER 消费失败，需要稍后重新消费  
                     return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;  
                 }  
             });  
      
             //调用start()方法启动consumer  
             consumer.start();  
            System.out.println("Consumer Started.");  
         }  
    }

(4)运行顺序：consumer1——consumer2——producer

consumer.setMessageModel(MessageModel.BROADCASTING);//设置广播消费模式

注意：为了使结果更直观，这边的ConsumerGroup设置为同一个

(5)测试结果

consumer1消费情况：

![70][]

consumer2消费情况：

![70 1][]

（6）与集群消费模式对比

默认的是使用集群消费模式，这两者最大的区别在于同组中的消费，集群消费模式是同组公同消费一组消息，广播模式是同组各自都消费一组消息。

下面看下集群消费模式的情况：

consumer1：

![70 2][]

consumer2：

![70 3][]

[70]: /images/20210923/1493076abb6d4bedb42a9ba4d075bed3.png
[70 1]: /images/20210923/261c7c94dd5d49e38daa8b064678592e.png
[70 2]: /images/20210923/28034ea315a34681b1f0f5cd43376817.png
[70 3]: /images/20210923/80b47cc25477402abd8dfddd6c0ff84c.png