久久免费视频2,九九热精品在线,国产高清在线视频

RocketMQTemplate發送帶tags的消息

RocketMQTemplate是RocketMQ集成到Spring cloud之后提供的個方便發送消息的模板類，它是基本Spring 的消息機制實現的，對外只提供了Spring抽象出來的消息發送接口。

在單獨使用RocketMQ的時候，發送消息使用的Message是‘org.apache.rocketmq.common.message'包下面的Message，而使用RocketMQTemplate發送消息時，使用的Message是org.springframework.messaging的Message，猛一看，沒辦法發送帶tags的消息了，其實在RocketMQ集成的時候已經解決了這個問題。

在RocketMQTemplate發送消息時，調用的方法是：

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();
//在這里對消息進行了轉化，將Spring的message轉化為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);
}
}

在上面的代碼中，對消息進行了轉化，將Spring的message轉化為rocketmq自己的message，在RocketMQUtil.convertToRocketMessage方法中有個地方就是獲取tags的：

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

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

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

使用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.RocketMQTemplate的bean，用它就可以直接把消息投遞出去。具體的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);

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

使用RocketMQTemplate發送帶tags的消息

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

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

使用RocketMQTemplate發送帶tags的消息

出錯的時候是這樣的

使用RocketMQTemplate發送帶tags的消息

由于普通消息沒有消息回查，普通消息用的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;
}
}