使用RocketMQTemplate發(fā)送帶tags的消息
RocketMQTemplate發(fā)送帶tags的消息
RocketMQTemplate是RocketMQ集成到Spring cloud之后提供的個(gè)方便發(fā)送消息的模板類，它是基本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增加依賴
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.0.1</version>
</dependency>
由于，這個(gè)依賴是一個(gè)starter，直接引入依賴就可以開(kāi)始寫投遞消息的代碼了。這個(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ò)異常，客戶端會(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的寫法，這個(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è)參考，也希望大家多多支持腳本之家。
相關(guān)文章
List是我們常用的集合，但是在多線程并發(fā)的條件下，會(huì)出現(xiàn)安全問(wèn)題嗎？下面我們就來(lái)測(cè)試一下，如果出現(xiàn)安全問(wèn)題，該如何解決，感興趣的可以了解一下2021-12-12
詳解JAVA后端實(shí)現(xiàn)統(tǒng)一掃碼支付：微信篇
本篇文章主要介紹了詳解JAVA后端實(shí)現(xiàn)統(tǒng)一掃碼支付：微信篇，這里整理了詳細(xì)的代碼,有需要的小伙伴可以參考下。2017-01-01
本文先對(duì)log4j進(jìn)行了簡(jiǎn)短的介紹，而后通過(guò)安裝、配置和普通項(xiàng)目和web項(xiàng)目幾個(gè)方面來(lái)詳細(xì)介紹了在Java下利用log4j記錄日志的方法，有需要的朋友們可以參考借鑒。2016-09-09
這篇文章主要介紹了快速理解Java設(shè)計(jì)模式中的組合模式，具有一定參考價(jià)值，需要的朋友可以了解下,希望能夠給你帶來(lái)幫助2021-09-09
基于注解的springboot+mybatis的多數(shù)據(jù)源組件的實(shí)現(xiàn)代碼
這篇文章主要介紹了基于注解的springboot+mybatis的多數(shù)據(jù)源組件的實(shí)現(xiàn),會(huì)使用到多個(gè)數(shù)據(jù)源，文中通過(guò)代碼講解的非常詳細(xì)，需要的朋友可以參考下2021-04-04
這篇文章主要介紹了JAVA類變量及類方法代碼實(shí)例詳解,文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2020-02-02
這篇文章主要為大家詳細(xì)介紹了java獲取微信accessToken的方法，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2019-09-09
Intellij IDEA 閱讀源碼的 4 個(gè)絕技(必看)
今天小編給大家分享Intellij IDEA 閱讀源碼的 4 個(gè)絕技，熟練的運(yùn)用 IDEA 中各個(gè)小技巧，讓閱讀跟蹤源碼變得更輕松，本文通過(guò)圖文并茂的形式給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友參考下吧2020-04-04
詳解J2EE開(kāi)發(fā)的網(wǎng)站部署到阿里云服務(wù)器的方法
這篇文章主要介紹了詳解J2EE開(kāi)發(fā)的網(wǎng)站部署到阿里云服務(wù)器的方法,需要的朋友可以參考下2018-01-01
利用logback 設(shè)置不同包下的日志級(jí)別
這篇文章主要介紹了利用logback 設(shè)置不同包下的日志級(jí)別，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2020-12-12