欧美一区二区三区老妇人-欧美做爰猛烈大尺度电-99久久夜色精品国产亚洲a-亚洲福利视频一区二区

RocketMQ怎么實現(xiàn)請求異步處理

這篇文章主要介紹“RocketMQ怎么實現(xiàn)請求異步處理”，在日常操作中，相信很多人在RocketMQ怎么實現(xiàn)請求異步處理問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”RocketMQ怎么實現(xiàn)請求異步處理”的疑惑有所幫助！接下來，請跟著小編一起來學習吧！

成都創(chuàng)新互聯(lián)提供成都網(wǎng)站設(shè)計、網(wǎng)站制作、網(wǎng)頁設(shè)計，品牌網(wǎng)站制作，廣告投放等致力于企業(yè)網(wǎng)站建設(shè)與公司網(wǎng)站制作，10年的網(wǎng)站開發(fā)和建站經(jīng)驗,助力企業(yè)信息化建設(shè)，成功案例突破上千,是您實現(xiàn)網(wǎng)站建設(shè)的好選擇.

一、RocketMQ

1、架構(gòu)圖片

RocketMQ怎么實現(xiàn)請求異步處理

2、角色分類

(1)、Broker

RocketMQ 的核心，接收 Producer 發(fā)過來的消息、處理 Consumer 的消費消息請求、消息的持 久化存儲、服務(wù)端過濾功能等 。

(2)、NameServer

消息隊列中的狀態(tài)服務(wù)器，集群的各個組件通過它來了解全局的信息 。類似微服務(wù)中注冊中心的服務(wù)注冊，發(fā)現(xiàn)，下線，上線的概念。

熱備份： NamServer可以部署多個，相互之間獨立，其他角色同時向多個NameServer 機器上報狀態(tài)信息。

心跳機制： NameServer 中的 Broker、 Topic等狀態(tài)信息不會持久存儲，都是由各個角色定時上報并存儲到內(nèi)存中，超時不上報的話， NameServer會認為某個機器出故障不可用。

(3)、Producer

消息的生成者，最常用的producer類就是DefaultMQProducer。

(4)、Consumer

消息的消費者，常用Consumer類 DefaultMQPushConsumer 收到消息后自動調(diào)用傳入的處理方法來處理，實時性高 DefaultMQPullConsumer 用戶自主控制 ，靈活性更高。

3、通信機制

(1)、Broker啟動后需要完成一次將自己注冊至NameServer的操作；隨后每隔30s時間定時向NameServer更新Topic路由信息。

(2)、Producer發(fā)送消息時候，需要根據(jù)消息的Topic從本地緩存的獲取路由信息。如果沒有則更新路由信息會從NameServer重新拉取，同時Producer會默認每隔30s向NameServer拉取一次路由信息。

(3)、Consumer消費消息時候，從NameServer獲取的路由信息，并再完成客戶端的負載均衡后，監(jiān)聽指定消息隊列獲取消息并進行消費。

二、代碼實現(xiàn)案例

1、項目結(jié)構(gòu)圖

RocketMQ怎么實現(xiàn)請求異步處理

版本描述

<spring-boot.version>2.1.3.RELEASE</spring-boot.version>
<rocketmq.version>4.3.0</rocketmq.version>

2、配置文件

rocketmq:
  # 生產(chǎn)者配置
  producer:
    isOnOff: on
    # 發(fā)送同一類消息的設(shè)置為同一個group，保證唯一
    groupName: CicadaGroup
    # 服務(wù)地址
    namesrvAddr: 127.0.0.1:9876
    # 消息最大長度 默認1024*4(4M)
    maxMessageSize: 4096
    # 發(fā)送消息超時時間,默認3000
    sendMsgTimeout: 3000
    # 發(fā)送消息失敗重試次數(shù)，默認2
    retryTimesWhenSendFailed: 2
  # 消費者配置
  consumer:
    isOnOff: on
    # 官方建議：確保同一組中的每個消費者訂閱相同的主題。
    groupName: CicadaGroup
    # 服務(wù)地址
    namesrvAddr: 127.0.0.1:9876
    # 接收該 Topic 下所有 Tag
    topics: CicadaTopic~*;
    consumeThreadMin: 20
    consumeThreadMax: 64
    # 設(shè)置一次消費消息的條數(shù)，默認為1條
    consumeMessageBatchMaxSize: 1

# 配置 Group  Topic  Tag
rocket:
  group: rocketGroup
  topic: rocketTopic
  tag: rocketTag

3、生產(chǎn)者配置

/**
 * RocketMQ 生產(chǎn)者配置
 */
@Configuration
public class ProducerConfig {

    private static final Logger LOG = LoggerFactory.getLogger(ProducerConfig.class) ;
    @Value("${rocketmq.producer.groupName}")
    private String groupName;
    @Value("${rocketmq.producer.namesrvAddr}")
    private String namesrvAddr;
    @Value("${rocketmq.producer.maxMessageSize}")
    private Integer maxMessageSize ;
    @Value("${rocketmq.producer.sendMsgTimeout}")
    private Integer sendMsgTimeout;
    @Value("${rocketmq.producer.retryTimesWhenSendFailed}")
    private Integer retryTimesWhenSendFailed;

    @Bean
    public DefaultMQProducer getRocketMQProducer() {
        DefaultMQProducer producer;
        producer = new DefaultMQProducer(this.groupName);
        producer.setNamesrvAddr(this.namesrvAddr);
        //如果需要同一個jvm中不同的producer往不同的mq集群發(fā)送消息，需要設(shè)置不同的instanceName
        if(this.maxMessageSize!=null){
            producer.setMaxMessageSize(this.maxMessageSize);
        }
        if(this.sendMsgTimeout!=null){
            producer.setSendMsgTimeout(this.sendMsgTimeout);
        }
        //如果發(fā)送消息失敗，設(shè)置重試次數(shù)，默認為2次
        if(this.retryTimesWhenSendFailed!=null){
            producer.setRetryTimesWhenSendFailed(this.retryTimesWhenSendFailed);
        }
        try {
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        return producer;
    }
}

4、消費者配置

/**
 * RocketMQ 消費者配置
 */
@Configuration
public class ConsumerConfig {
    private static final Logger LOG = LoggerFactory.getLogger(ConsumerConfig.class) ;
    @Value("${rocketmq.consumer.namesrvAddr}")
    private String namesrvAddr;
    @Value("${rocketmq.consumer.groupName}")
    private String groupName;
    @Value("${rocketmq.consumer.consumeThreadMin}")
    private int consumeThreadMin;
    @Value("${rocketmq.consumer.consumeThreadMax}")
    private int consumeThreadMax;
    @Value("${rocketmq.consumer.topics}")
    private String topics;
    @Value("${rocketmq.consumer.consumeMessageBatchMaxSize}")
    private int consumeMessageBatchMaxSize;
    @Resource
    private RocketMsgListener msgListener;
    @Bean
    public DefaultMQPushConsumer getRocketMQConsumer(){
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setConsumeThreadMin(consumeThreadMin);
        consumer.setConsumeThreadMax(consumeThreadMax);
        consumer.registerMessageListener(msgListener);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
        try {
            String[] topicTagsArr = topics.split(";");
            for (String topicTags : topicTagsArr) {
                String[] topicTag = topicTags.split("~");
                consumer.subscribe(topicTag[0],topicTag[1]);
            }
            consumer.start();
        }catch (MQClientException e){
            e.printStackTrace();
        }
        return consumer;
    }
}

5、消息監(jiān)聽配置

/**
 * 消息消費監(jiān)聽
 */
@Component
public class RocketMsgListener implements MessageListenerConcurrently {
    private static final Logger LOG = LoggerFactory.getLogger(RocketMsgListener.class) ;
    @Resource
    private ParamConfigService paramConfigService ;
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
        if (CollectionUtils.isEmpty(list)){
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
        MessageExt messageExt = list.get(0);
        LOG.info("接受到的消息為："+new String(messageExt.getBody()));
        int reConsume = messageExt.getReconsumeTimes();
        // 消息已經(jīng)重試了3次，如果不需要再次消費，則返回成功
        if(reConsume ==3){
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
        if(messageExt.getTopic().equals(paramConfigService.rocketTopic)){
            String tags = messageExt.getTags() ;
            switch (tags){
                case "rocketTag":
                    LOG.info("開戶 tag == >>"+tags);
                    break ;
                default:
                    LOG.info("未匹配到Tag == >>"+tags);
                    break;
            }
        }
        // 消息消費成功
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}

6、配置參數(shù)綁定

@Service
public class ParamConfigService {
    @Value("${rocket.group}")
    public String rocketGroup ;
    @Value("${rocket.topic}")
    public String rocketTopic ;
    @Value("${rocket.tag}")
    public String rocketTag ;
}

7、消息發(fā)送測試

@Service
public class RocketMqServiceImpl implements RocketMqService {
    @Resource
    private DefaultMQProducer defaultMQProducer;
    @Resource
    private ParamConfigService paramConfigService ;
    @Override
    public SendResult openAccountMsg(String msgInfo) {
        // 可以不使用Config中的Group
        defaultMQProducer.setProducerGroup(paramConfigService.rocketGroup);
        SendResult sendResult = null;
        try {
            Message sendMsg = new Message(paramConfigService.rocketTopic,
                                          paramConfigService.rocketTag,
                                         "open_account_key", msgInfo.getBytes());
            sendResult = defaultMQProducer.send(sendMsg);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return sendResult ;
    }
}

三、項目源碼

GitHub·地址
https://github.com/cicadasmile/middle-ware-parent
GitEE·地址
https://gitee.com/cicadasmile/middle-ware-parent

到此，關(guān)于“RocketMQ怎么實現(xiàn)請求異步處理”的學習就結(jié)束了，希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學習，快去試試吧！若想繼續(xù)學習更多相關(guān)知識，請繼續(xù)關(guān)注創(chuàng)新互聯(lián)網(wǎng)站，小編會繼續(xù)努力為大家?guī)砀鄬嵱玫奈恼拢?/p>

當前題目：RocketMQ怎么實現(xiàn)請求異步處理
網(wǎng)站鏈接：http://chinadenli.net/article6/gsgsog.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供用戶體驗、網(wǎng)頁設(shè)計公司、手機網(wǎng)站建設(shè)、微信小程序、定制開發(fā)、域名注冊

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時需注明來源： 創(chuàng)新互聯(lián)

手機網(wǎng)站建設(shè)