91超碰碰碰碰久久久久久综合_超碰av人澡人澡人澡人澡人掠_国产黄大片在线观看画质优化_txt小说免费全本

溫馨提示×

溫馨提示×

您好,登錄后才能下訂單哦!

密碼登錄×
登錄注冊×
其他方式登錄
點擊 登錄注冊 即表示同意《億速云用戶服務條款》

使用RocketMQ怎么對消息進行處理

發布時間:2021-06-15 11:59:23 來源:億速云 閱讀:249 作者:Leah 欄目:大數據

這期內容當中小編將會給大家帶來有關使用RocketMQ怎么對消息進行處理,文章內容豐富且以專業的角度為大家分析和敘述,閱讀完這篇文章希望大家可以有所收獲。

消息發送(生產者)

以maven + SpringBoot 工程為例,先在pom.xml增加依賴

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.0.1</version>
</dependency>

由于,這個依賴是一個starter,直接引入依賴就可以開始寫投遞消息的代碼了。這個starter注冊了一個叫org.apache.rocketmq.spring.core.RocketMQTemplatebean,用它就可以直接把消息投遞出去。 具體的API是這樣的

    XXXEvent xxxDto = new XXXEvent();
    Message<XXXEvent> message = MessageBuilder.withPayload(xxxDto).build();
    String dest = String.format("%s:%s",topic-name","tag-name");
    //默認投遞:同步發送 不會丟失消息。如果在投遞成功后發生網絡異常,客戶端會認為投遞失敗而回滾本地事務
    this.rocketMQTemplate.send(dest, xxxDto);

這種投遞方式能保證投遞成功的消息不會丟失,但是不能保證投遞一定成功。假設一次調用的流程是這樣的

使用RocketMQ怎么對消息進行處理

如果在步驟3的時候發生錯誤,因為出錯mqClient會認為消息投遞失敗而把事務回滾。如果消息已經被消費,那就會導致業務錯誤。我們可以用事務消息解決這個問題

以帶事務方式投遞的消息,正常情況下的處理流程是這樣的

使用RocketMQ怎么對消息進行處理

出錯的時候是這樣的

使用RocketMQ怎么對消息進行處理

由于普通消息沒有消息回查,普通消息用的producer不支持回查操作,不同業務的回查處理也不一樣,事務消息需要使用單獨的producer。消息發送代碼大概是這樣的

//調用這段代碼之前別做會影響數據的操作
XXXEvent xxxDto = new XXXEvent();
Message<XXXEvent> message = MessageBuilder.withPayload(xxxDto).build();
String dest = String.format("%s:%s",topic-name","tag-name");
TransactionSendResult transactionSendResult = this.rocketMQTemplate.sendMessageInTransaction("poducer-name","topic-name:tag-name",message,"xxxid");
if (LocalTransactionState.ROLLBACK_MESSAGE.equals(transactionSendResult.getLocalTransactionState())){
    throw new RuntimeException("事務消息投遞失敗");
}
//按照RocketMQ的寫法,這個地方不應該有別的代碼
@RocketMQTransactionListener(txProducerGroup = "producer")
    class TransactionListenerImpl implements RocketMQLocalTransactionListener {
        
        //消息投遞成功后執行的邏輯(半消息)
        //原文:When send transactional prepare(half) message succeed, this method will be invoked to execute local transaction.
        @Override
        public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
            try{
                //
                xxxService.doSomething();
                return RocketMQLocalTransactionState.COMMIT;
            catch(IOException e){
                //不確定最終是否成功
                return RocketMQLocalTransactionState.UNKNOWN;
            }catch(Exception e){
                return RocketMQLocalTransactionState.ROLLBACK;
            }
        }

        //回查事務執行狀態
        @Override
        public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
            Boolean result = xxxService.isSuccess(msg,arg);
            if(result != null){
                if(result){
                    return RocketMQLocalTransactionState.COMMIT;
                }else{
                    return RocketMQLocalTransactionState.ROLLBACK;
                }
            }
            return RocketMQLocalTransactionState.UNKNOWN;
        }
    }
處理消息(消費)

普通消息和事務消息的區別只在投遞的時候才明顯,對應的消費端代碼比較簡單

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;


@Slf4j
@Component
@RocketMQMessageListener(consumerGroup = "xxx-consumer", topic = "topic-name",selectorExpression = "tag-name")
public class XXXEventMQListener implements RocketMQListener<XXXEvent> {
    private  String repeatCheckRedisKeyTemplate = "topic-name:tag:repeat-check:%s";

    @Autowired private StringRedisTemplate redisTemplate;

    @Override
    public void onMessage(XXXEvent message) {
        log.info("consumer message {}",message);
        //處理消息
        try{
            xxxService.doSomething(message);
        }catch(Exception ex){
            log.warn(String.format("message [%s] 消費失敗",message),ex);
            //拋出異常后,MQClient會返回ConsumeConcurrentlyStatus.RECONSUME_LATER,這條消息會再次嘗試消費
            throw new RuntimException(ex);
        }
    }
}

RocketMQ用ACK機制保證NameServer知道消息是否被消費org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer里是這么處理的

public class DefaultMessageListenerConcurrently implements MessageListenerConcurrently {

    @SuppressWarnings("unchecked")
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        for (MessageExt messageExt : msgs) {
            log.debug("received msg: {}", messageExt);
            try {
                long now = System.currentTimeMillis();
                rocketMQListener.onMessage(doConvertMessage(messageExt));
                long costTime = System.currentTimeMillis() - now;
                log.debug("consume {} cost: {} ms", messageExt.getMsgId(), costTime);
            } catch (Exception e) {
                log.warn("consume message failed. messageExt:{}", messageExt, e);
                context.setDelayLevelWhenNextConsume(delayLevelWhenNextConsume);
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        }

        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}

上述就是小編為大家分享的使用RocketMQ怎么對消息進行處理了,如果剛好有類似的疑惑,不妨參照上述分析進行理解。如果想知道更多相關知識,歡迎關注億速云行業資訊頻道。

向AI問一下細節

免責聲明:本站發布的內容(圖片、視頻和文字)以原創、轉載和分享為主,文章觀點不代表本網站立場,如果涉及侵權請聯系站長郵箱:is@yisu.com進行舉報,并提供相關證據,一經查實,將立刻刪除涉嫌侵權內容。

AI

桐城市| 双峰县| 屏南县| 普兰县| 资溪县| 安庆市| 科技| 揭阳市| 龙南县| 天镇县| 旬邑县| 莆田市| 尼木县| 大名县| 霍邱县| 安多县| 陇西县| 邳州市| 长汀县| 沁阳市| 波密县| 大姚县| 崇文区| 吉木萨尔县| 宁国市| 扎赉特旗| 盘锦市| 康平县| 呈贡县| 凤庆县| 洪江市| 南城县| 阿鲁科尔沁旗| 义马市| 仙居县| 阜新| 三穗县| 灌南县| 丰宁| 邓州市| 石嘴山市|