您好,登錄后才能下訂單哦!
這篇文章將為大家詳細(xì)講解有關(guān)如何使用RocketMQTemplate發(fā)送帶tags的消息,小編覺得挺實(shí)用的,因此分享給大家做個參考,希望大家閱讀完這篇文章后可以有所收獲。
RocketMQTemplate是RocketMQ集成到Spring cloud之后提供的個方便發(fā)送消息的模板類,它是基本Spring 的消息機(jī)制實(shí)現(xiàn)的,對外只提供了Spring抽象出來的消息發(fā)送接口。
在單獨(dú)使用RocketMQ的時(shí)候,發(fā)送消息使用的Message是‘org.apache.rocketmq.common.message'包下面的Message,而使用RocketMQTemplate發(fā)送消息時(shí),使用的Message是org.springframework.messaging的Message,猛一看,沒辦法發(fā)送帶tags的消息了,其實(shí)在RocketMQ集成的時(shí)候已經(jīng)解決了這個問題。
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(); //在這里對消息進(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); } }
在上面的代碼中,對消息進(jìn)行了轉(zhuǎn)化,將Spring的message轉(zhuǎn)化為rocketmq自己的message,在RocketMQUtil.convertToRocketMessage方法中有個地方就是獲取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
以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.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ā)送 不會丟失消息。如果在投遞成功后發(fā)生網(wǎng)絡(luò)異常,客戶端會認(rèn)為投遞失敗而回滾本地事務(wù) this.rocketMQTemplate.send(dest, xxxDto);
這種投遞方式能保證投遞成功的消息不會丟失,但是不能保證投遞一定成功。假設(shè)一次調(diào)用的流程是這樣的
如果在步驟3的時(shí)候發(fā)生錯誤,因?yàn)槌鲥emqClient會認(rèn)為消息投遞失敗而把事務(wù)回滾。如果消息已經(jīng)被消費(fèi),那就會導(dǎo)致業(yè)務(wù)錯誤。我們可以用事務(wù)消息解決這個問題。
以帶事務(wù)方式投遞的消息,正常情況下的處理流程是這樣的
出錯的時(shí)候是這樣的
由于普通消息沒有消息回查,普通消息用的producer不支持回查操作,不同業(yè)務(wù)的回查處理也不一樣,事務(wù)消息需要使用單獨(dú)的producer。消息發(fā)送代碼大概是這樣的
//調(diào)用這段代碼之前別做會影響數(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的寫法,這個地方不應(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; } }
普通消息和事務(wù)消息的區(qū)別只在投遞的時(shí)候才明顯,對應(yīng)的消費(fèi)端代碼比較簡單
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會返回ConsumeConcurrentlyStatus.RECONSUME_LATER,這條消息會再次嘗試消費(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; } }
關(guān)于“如何使用RocketMQTemplate發(fā)送帶tags的消息”這篇文章就分享到這里了,希望以上內(nèi)容可以對大家有一定的幫助,使各位可以學(xué)到更多知識,如果覺得文章不錯,請把它分享出去讓更多的人看到。
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場,如果涉及侵權(quán)請聯(lián)系站長郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。