RocketMQ的兩種消費(fèi)模式詳解

更新時(shí)間：2023年10月11日 11:07:35 作者：碼奴生來只知道前進(jìn)~

這篇文章主要介紹了RocketMQ的兩種消費(fèi)模式詳解,RocketMQ主要提供了兩種消費(fèi)模式,集群消費(fèi)以及廣播消費(fèi),我們只需要在定義消費(fèi)者的時(shí)候通過setMessageModel(MessageModel.XXX),需要的朋友可以參考下

1、添加依賴

<dependency>
	<groupId>org.apache.rocketmq</groupId>
	<artifactId>rocketmq-client</artifactId>
	<version>4.4.0</version>
</dependency>
<dependency>
	<groupId>com.alibaba</groupId>
	<artifactId>fastjson</artifactId>
	<version>1.2.3</version>
</dependency>
<dependencies>
	<dependency>
		<groupId>org.projectlombok</groupId>
		<artifactId>lombok</artifactId>
	</dependency>
</dependencies>

2、消費(fèi)模式

RocketMQ主要提供了兩種消費(fèi)模式：集群消費(fèi)以及廣播消費(fèi)。我們只需要在定義消費(fèi)者的時(shí)候通過setMessageModel(MessageModel.XXX)

// 設(shè)置消費(fèi)模型，集群還是廣播，默認(rèn)為集群  CLUSTERING-集群，BROADCASTING-廣播
mqPushConsumer.setMessageModel(MessageModel.CLUSTERING);
mqPushConsumer.setMessageModel(MessageModel.BROADCASTING);

方法就可以指定是集群還是廣播式消費(fèi)，默認(rèn)是集群消費(fèi)模式，即每個(gè)Consumer Group中的Consumer均攤所有的消息。

3、集群消費(fèi)

3.1 生產(chǎn)者

package com.shucha.deveiface.biz.mq.producer;
import com.alibaba.fastjson.JSON;
import com.shucha.deveiface.biz.model.User;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import java.util.Date;
/**
 * @author tqf
 * @Description 生產(chǎn)者
 * @Version 1.0
 * @since 2022-07-12 14:50
 */
public class MQProducer {
    public static void main(String[] args) throws MQClientException{
        producerSendMessage();
    }
    /**
     * 生產(chǎn)消息方法
     * @throws MQClientException
     */
    public static void producerSendMessage() throws MQClientException {
        // 創(chuàng)建DefaultMQProducer類并設(shè)定生產(chǎn)者名稱
        DefaultMQProducer mqProducer = new DefaultMQProducer("producer-group-test");
        // 設(shè)置NameServer地址，如果是集群的話，使用分號(hào);分隔開
        mqProducer.setNamesrvAddr("127.0.0.1:9876");
        // 消息最大長度 默認(rèn)4M
        mqProducer.setMaxMessageSize(4096);
        // 發(fā)送消息超時(shí)時(shí)間，默認(rèn)3000
        mqProducer.setSendMsgTimeout(3000);
        // 發(fā)送消息失敗重試次數(shù)，默認(rèn)2
        mqProducer.setRetryTimesWhenSendAsyncFailed(2);
        // 啟動(dòng)消息生產(chǎn)者
        mqProducer.start();
        try {
            // 循環(huán)十次，發(fā)送十條消息
            for (int i = 1; i <= 10; i++) {
                User user = new User();
                user.setId((long)i);
                user.setAge(i);
                user.setUserName("姓名"+i);
                user.setCreateTime(new Date());
                // String msg = "這是第" + i + "條消息測試";
                String msg = JSON.toJSONString(user);
                // 創(chuàng)建消息，并指定Topic(主題)，Tag(標(biāo)簽)和消息內(nèi)容
                Message message = new Message("TOPIC_TEST", "", msg.getBytes(RemotingHelper.DEFAULT_CHARSET));
                // 發(fā)送同步消息到一個(gè)Broker，可以通過sendResult返回消息是否成功送達(dá)
                SendResult sendResult = mqProducer.send(message);
                // mqProducer.sendOneway(message);
                // 消息id
                /*System.out.println(sendResult.getMsgId());
                // 隊(duì)列信息
                System.out.println(sendResult.getMessageQueue());
                // 發(fā)送結(jié)果
                System.out.println(sendResult.getSendStatus());
                // 下一個(gè)要消費(fèi)的消息的偏移量
                System.out.println(sendResult.getOffsetMsgId());
                // 隊(duì)列消息偏移量
                System.out.println(sendResult.getQueueOffset());*/
                System.out.println(sendResult);
            }
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("生產(chǎn)消息異常！");
        }
        // 如果不再發(fā)送消息，關(guān)閉Producer實(shí)例
        mqProducer.shutdown();
    }
}

User用戶測試實(shí)體類

package com.shucha.deveiface.biz.model;
import com.fasterxml.jackson.annotation.JsonFormat;
import com.sdy.common.utils.DateUtil;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
import java.util.Date;
/**
 * @author tqf
 * @Description
 * @Version 1.0
 * @since 2022-04-07 13:53
 */
@Data
public class User {
    /**
     * 主鍵ID
     */
    private Long id;
    /**
     *用戶名
     */
    private String userName;
    /**
     * 用戶密碼
     */
    private String passWord;
    /**
     * 年齡
     */
    private Integer age;
    /**
     * 性別(0-男,1-女,2-未知)
     */
    private Integer sex;
    /**
     * 創(chuàng)建時(shí)間
     */
    @JsonFormat(pattern = DateUtil.DATETIME_FORMAT)
    private Date createTime;
}

3.2 消費(fèi)者A

package com.shucha.deveiface.biz.mq.consumer;
import com.alibaba.fastjson.JSON;
import com.shucha.deveiface.biz.constants.MqConstants;
import com.shucha.deveiface.biz.model.User;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import java.nio.charset.Charset;
import java.util.List;
/**
 * @author tqf
 * @Description 消費(fèi)者A
 * @Version 1.0
 * @since 2022-07-12 14:37
 */
public class ConsumerA {
    public static void main(String[] args) throws MQClientException {
        ConsumerA();
    }
    public static void ConsumerA() throws MQClientException {
        // 創(chuàng)建DefaultMQPushConsumer類并設(shè)定消費(fèi)者名稱
        DefaultMQPushConsumer mqPushConsumer = new DefaultMQPushConsumer(MqConstants.ConsumerGroup.CONSUMER_GROUP1);
        // DefaultMQPullConsumer pullConsumer = new DefaultMQPullConsumer(MqConstants.ConsumerGroup.CONSUMER_GROUP1);
        // 設(shè)置NameServer地址，如果是集群的話，使用分號(hào);分隔開
        mqPushConsumer.setNamesrvAddr("127.0.0.1:9876");
        // pullConsumer.setNamesrvAddr("127.0.0.1:9876");
        // 設(shè)置Consumer第一次啟動(dòng)是從隊(duì)列頭部開始消費(fèi)還是隊(duì)列尾部開始消費(fèi)
        // 如果不是第一次啟動(dòng)，那么按照上次消費(fèi)的位置繼續(xù)消費(fèi)
        mqPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // 設(shè)置消費(fèi)模型，集群還是廣播，默認(rèn)為集群  CLUSTERING-集群   BROADCASTING-廣播
        mqPushConsumer.setMessageModel(MessageModel.CLUSTERING);
        // mqPushConsumer.setMessageModel(MessageModel.BROADCASTING);
        // 消費(fèi)者最小線程量
        mqPushConsumer.setConsumeThreadMin(5);
        // 消費(fèi)者最大線程量
        mqPushConsumer.setConsumeThreadMax(10);
        // 設(shè)置一次消費(fèi)消息的條數(shù)，默認(rèn)是1
        mqPushConsumer.setConsumeMessageBatchMaxSize(1);
        // 訂閱一個(gè)或者多個(gè)Topic，以及Tag來過濾需要消費(fèi)的消息，如果訂閱該主題下的所有tag，則使用*
        mqPushConsumer.subscribe("TOPIC_TEST", "*");
        // 注冊(cè)回調(diào)實(shí)現(xiàn)類來處理從broker拉取回來的消息
        mqPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            // 監(jiān)聽類實(shí)現(xiàn)MessageListenerConcurrently接口即可，重寫consumeMessage方法接收數(shù)據(jù)
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgList, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (MessageExt msg : msgList) {
                    String msgBody = new String(msg.getBody(), Charset.forName(RemotingHelper.DEFAULT_CHARSET));
                    System.out.println("消費(fèi)者A接收到消息：" +  "===== " + msgBody);
                    /*User user = JSON.parseObject(msgBody, User.class);
                    System.out.println("消費(fèi)者A接收到消息：" +  "===== " + user.getId());*/
                }
                /*MessageExt messageExt = msgList.get(0);
                String body = new String(messageExt.getBody(), StandardCharsets.UTF_8);
                System.out.println("消費(fèi)者A接收到消息: " + messageExt.toString() + "---消息內(nèi)容為：" + body);*/
                // 標(biāo)記該消息已經(jīng)被成功消費(fèi)
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 啟動(dòng)消費(fèi)者實(shí)例
        mqPushConsumer.start();
        System.out.println("ConsumerA Started.");
    }
}

3.3 消費(fèi)者B

package com.shucha.deveiface.biz.mq.consumer;
import com.shucha.deveiface.biz.constants.MqConstants;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import java.util.List;
/**
 * @author tqf
 * @Description 消費(fèi)者B
 * @Version 1.0
 * @since 2022-07-12 14:40
 */
public class ConsumerB {
    public static void main(String[] args) throws MQClientException {
        ConsumerB();
    }
    public static void ConsumerB() throws MQClientException {
        // 創(chuàng)建DefaultMQPushConsumer類并設(shè)定消費(fèi)者名稱
        DefaultMQPushConsumer mqPushConsumer = new DefaultMQPushConsumer(MqConstants.ConsumerGroup.CONSUMER_GROUP1);
        // 設(shè)置NameServer地址，如果是集群的話，使用分號(hào);分隔開
        mqPushConsumer.setNamesrvAddr("127.0.0.1:9876");
        // 設(shè)置Consumer第一次啟動(dòng)是從隊(duì)列頭部開始消費(fèi)還是隊(duì)列尾部開始消費(fèi)
        // 如果不是第一次啟動(dòng)，那么按照上次消費(fèi)的位置繼續(xù)消費(fèi)
        mqPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // 設(shè)置消費(fèi)模型，集群還是廣播，默認(rèn)為集群  CLUSTERING-集群   BROADCASTING-廣播
        // mqPushConsumer.setMessageModel(MessageModel.CLUSTERING);
        mqPushConsumer.setMessageModel(MessageModel.BROADCASTING);
        // 消費(fèi)者最小線程量
        mqPushConsumer.setConsumeThreadMin(5);
        // 消費(fèi)者最大線程量
        mqPushConsumer.setConsumeThreadMax(10);
        // 設(shè)置一次消費(fèi)消息的條數(shù)，默認(rèn)是1
        mqPushConsumer.setConsumeMessageBatchMaxSize(1);
        // 訂閱一個(gè)或者多個(gè)Topic，以及Tag來過濾需要消費(fèi)的消息，如果訂閱該主題下的所有tag，則使用*
        mqPushConsumer.subscribe("TOPIC_TEST", "*");
        // 注冊(cè)回調(diào)實(shí)現(xiàn)類來處理從broker拉取回來的消息
        mqPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            // 監(jiān)聽類實(shí)現(xiàn)MessageListenerConcurrently接口即可，重寫consumeMessage方法接收數(shù)據(jù)
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgList, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
               /* MessageExt messageExt = msgList.get(0);
                String body = new String(messageExt.getBody(), StandardCharsets.UTF_8);
                System.out.println("消費(fèi)者B接收到消息: " + messageExt.toString() + "---消息內(nèi)容為：" + body);*/
                for (MessageExt msg : msgList) {
                    System.out.println("消費(fèi)者B接收到消息：" +  "===== " + new String(msg.getBody()));
                    // String msgBody = new String(msg.getBody(), Charset.forName(RemotingHelper.DEFAULT_CHARSET));
                }
                // 標(biāo)記該消息已經(jīng)被成功消費(fèi)
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 啟動(dòng)消費(fèi)者實(shí)例
        mqPushConsumer.start();
        System.out.println("ConsumerB Started.");
    }
}

可以看到，生產(chǎn)者發(fā)送了10條消息，ConsumerA與ConsumerB屬于同一個(gè)消費(fèi)者組，集群消費(fèi)模式下每個(gè)消費(fèi)者攤分消費(fèi)所有消息。注意，兩個(gè)消費(fèi)者的ConsumerGroup組名需要一致，才算是同一個(gè)消費(fèi)者組。

簡單總結(jié)一下：

1、在Rocket集群消費(fèi)模式下，（訂閱）同一個(gè)主題（Topic）下的消息，對(duì)于不同的消費(fèi)者組是一種“廣播形式”，即每個(gè)消費(fèi)者組的都會(huì)消費(fèi)消息。

2、在Rocket集群消費(fèi)模式下，（訂閱）同一個(gè)主題（Topic）下的消息，對(duì)于相同的消費(fèi)者組的消費(fèi)者而言是一種集群模式，即同一個(gè)消費(fèi)者組內(nèi)的所有消費(fèi)者均分消息并消費(fèi)。

4、廣播消費(fèi)

一條消息被多個(gè) Consumer 消費(fèi)，即使這些 Consumer 屬于同一個(gè) Consumer Group，消息也會(huì)被 Consumer Group 中的每個(gè) Consumer 都消費(fèi)一次，廣播消費(fèi)中的 Consumer Group 概念可以認(rèn)為在消息劃分方面無意義。

使用方法：setMessageModel(MessageModel.BROADCASTING)

將前面的消費(fèi)者A和消費(fèi)者B的集群模式代碼設(shè)置為如下

mqPushConsumer.setMessageModel(MessageModel.BROADCASTING);

重新啟動(dòng)生成者和2個(gè)消費(fèi)者

4.1 生產(chǎn)者數(shù)據(jù)

4.2 消費(fèi)者A

4.3 消費(fèi)者B

可以看到生產(chǎn)者發(fā)送了10條消息，ConsumerA與ConsumerB屬于同一個(gè)消費(fèi)者組，廣播模式下每個(gè)消費(fèi)者都會(huì)全量消費(fèi)所有消息。

集群消費(fèi)：任何一條消息只需要被消費(fèi)者集群中任意一個(gè)消費(fèi)者處理。
廣播消費(fèi)：每條消息被推送給消費(fèi)者集群中的所有注冊(cè)消費(fèi)者，保證消息被每個(gè)消費(fèi)者至少消費(fèi)一次。

到此這篇關(guān)于RocketMQ的兩種消費(fèi)模式詳解的文章就介紹到這了,更多相關(guān)RocketMQ消費(fèi)模式內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章:

国产无遮挡裸体免费直播视频,久久精品国产蜜臀av,动漫在线视频一区二区,欧亚日韩一区二区三区,久艹在线免费视频,国产精品美女网站免费,正在播放 97超级视频在线观看,斗破苍穹年番在线观看免费,51最新乱码中文字幕

RocketMQ的兩種消費(fèi)模式詳解

目錄

1、添加依賴

2、消費(fèi)模式

3、集群消費(fèi)

3.1 生產(chǎn)者

3.2 消費(fèi)者A

3.3 消費(fèi)者B

4、廣播消費(fèi)

4.1 生產(chǎn)者數(shù)據(jù)

4.2 消費(fèi)者A

4.3 消費(fèi)者B

相關(guān)文章

最新評(píng)論

大家感興趣的內(nèi)容

最近更新的內(nèi)容

常用在線小工具

国产无遮挡裸体免费直播视频,久久精品国产蜜臀av,动漫在线视频一区二区,欧亚日韩一区二区三区,久艹在线 免费视频,国产精品美女网站免费,正在播放 97超级视频在线观看,斗破苍穹年番在线观看免费,51最新乱码中文字幕

RocketMQ的兩種消費(fèi)模式詳解

目錄

1、添加依賴

2、消費(fèi)模式

3、集群消費(fèi)

3.1 生產(chǎn)者

3.2 消費(fèi)者A

3.3 消費(fèi)者B

4、廣播消費(fèi)

4.1 生產(chǎn)者數(shù)據(jù)

4.2 消費(fèi)者A

4.3 消費(fèi)者B

相關(guān)文章

最新評(píng)論

大家感興趣的內(nèi)容

最近更新的內(nèi)容

常用在線小工具

国产无遮挡裸体免费直播视频,久久精品国产蜜臀av,动漫在线视频一区二区,欧亚日韩一区二区三区,久艹在线免费视频,国产精品美女网站免费,正在播放 97超级视频在线观看,斗破苍穹年番在线观看免费,51最新乱码中文字幕

1、添加依賴

2、消費(fèi)模式