使用RocketMQTemplate發(fā)送帶tags的消息

更新時(shí)間：2021年07月02日 11:01:08 作者：wgslucky

這篇文章主要介紹了使用RocketMQTemplate發(fā)送帶tags的消息，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教

RocketMQTemplate發(fā)送帶tags的消息

RocketMQTemplate是RocketMQ集成到Spring cloud之后提供的個(gè)方便發(fā)送消息的模板類(lèi)，它是基本Spring 的消息機(jī)制實(shí)現(xiàn)的，對(duì)外只提供了Spring抽象出來(lái)的消息發(fā)送接口。

在單獨(dú)使用RocketMQ的時(shí)候，發(fā)送消息使用的Message是‘org.apache.rocketmq.common.message'包下面的Message，而使用RocketMQTemplate發(fā)送消息時(shí)，使用的Message是org.springframework.messaging的Message，猛一看，沒(méi)辦法發(fā)送帶tags的消息了，其實(shí)在RocketMQ集成的時(shí)候已經(jīng)解決了這個(gè)問(wèn)題。

在RocketMQTemplate發(fā)送消息時(shí)，調(diào)用的方法是：

public SendResult syncSendOrderly(String destination, Message<?> message, String hashKey, long timeout) {
        if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {
            log.error("syncSendOrderly failed. destination:{}, message is null ", destination);
            throw new IllegalArgumentException("`message` and `message.payload` cannot be null");
        }
        try {
            long now = System.currentTimeMillis();
          //在這里對(duì)消息進(jìn)行了轉(zhuǎn)化，將Spring的message轉(zhuǎn)化為rocketmq自己的message
            org.apache.rocketmq.common.message.Message rocketMsg = RocketMQUtil.convertToRocketMessage(objectMapper,
                charset, destination, message);
            SendResult sendResult = producer.send(rocketMsg, messageQueueSelector, hashKey, timeout);
            long costTime = System.currentTimeMillis() - now;
            log.debug("send message cost: {} ms, msgId:{}", costTime, sendResult.getMsgId());
            return sendResult;
        } catch (Exception e) {
            log.error("syncSendOrderly failed. destination:{}, message:{} ", destination, message);
            throw new MessagingException(e.getMessage(), e);
        }
    }

在上面的代碼中，對(duì)消息進(jìn)行了轉(zhuǎn)化，將Spring的message轉(zhuǎn)化為rocketmq自己的message，在RocketMQUtil.convertToRocketMessage方法中有個(gè)地方就是獲取tags的：

        String[] tempArr = destination.split(":", 2);
        String topic = tempArr[0];
        String tags = "";
        if (tempArr.length > 1) {
            tags = tempArr[1];
        }

所以，在發(fā)送消息的時(shí)候，我們只要把tags使用":"添加到topic后面就可以了。

例如：xxxx:tag1 || tag2 || tag3

使用RocketMQ 處理消息

消息發(fā)送（生產(chǎn)者）

以maven + SpringBoot 工程為例，先在pom.xml增加依賴(lài)

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

由于，這個(gè)依賴(lài)是一個(gè)starter，直接引入依賴(lài)就可以開(kāi)始寫(xiě)投遞消息的代碼了。這個(gè)starter注冊(cè)了一個(gè)叫org.apache.rocketmq.spring.core.RocketMQTemplate的bean，用它就可以直接把消息投遞出去。具體的API是這樣的

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

這種投遞方式能保證投遞成功的消息不會(huì)丟失，但是不能保證投遞一定成功。假設(shè)一次調(diào)用的流程是這樣的

如果在步驟3的時(shí)候發(fā)生錯(cuò)誤，因?yàn)槌鲥e(cuò)mqClient會(huì)認(rèn)為消息投遞失敗而把事務(wù)回滾。如果消息已經(jīng)被消費(fèi)，那就會(huì)導(dǎo)致業(yè)務(wù)錯(cuò)誤。我們可以用事務(wù)消息解決這個(gè)問(wèn)題。

以帶事務(wù)方式投遞的消息，正常情況下的處理流程是這樣的

出錯(cuò)的時(shí)候是這樣的

由于普通消息沒(méi)有消息回查，普通消息用的producer不支持回查操作，不同業(yè)務(wù)的回查處理也不一樣，事務(wù)消息需要使用單獨(dú)的producer。消息發(fā)送代碼大概是這樣的

//調(diào)用這段代碼之前別做會(huì)影響數(shù)據(jù)的操作
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("事務(wù)消息投遞失敗");
}
//按照RocketMQ的寫(xiě)法，這個(gè)地方不應(yīng)該有別的代碼

@RocketMQTransactionListener(txProducerGroup = "producer")
    class TransactionListenerImpl implements RocketMQLocalTransactionListener {
        
        //消息投遞成功后執(zhí)行的邏輯(半消息)
        //原文: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;
            }
        }
        //回查事務(wù)執(zhí)行狀態(tài)
        @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;
        }
    }

處理消息（消費(fèi)）

普通消息和事務(wù)消息的區(qū)別只在投遞的時(shí)候才明顯，對(duì)應(yīng)的消費(fèi)端代碼比較簡(jiǎn)單

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] 消費(fèi)失敗",message),ex);
            //拋出異常后，MQClient會(huì)返回ConsumeConcurrentlyStatus.RECONSUME_LATER,這條消息會(huì)再次嘗試消費(fèi)
            throw new RuntimException(ex);
        }
    }
}

RocketMQ用ACK機(jī)制保證NameServer知道消息是否被消費(fèi)在

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;
    }
}

以上為個(gè)人經(jīng)驗(yàn)，希望能給大家一個(gè)參考，也希望大家多多支持腳本之家。

您可能感興趣的文章: