package com.gkhy.safePlatform.safeCheck.common; import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.client.producer.SendStatus; import org.apache.rocketmq.spring.core.RocketMQTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import org.springframework.transaction.support.TransactionSynchronizationAdapter; import org.springframework.transaction.support.TransactionSynchronizationManager; @Component public class RocketMQTemplateHelper { @Autowired private RocketMQTemplate rocketMQTemplate; /** * 事务提交后发送MQ * @param message * @param */ public void syncSend(String destination, T message) { // 是否开启事务判断 if (TransactionSynchronizationManager.isSynchronizationActive()) { TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() { @Override public void afterCommit() { SendResult sendResult = rocketMQTemplate.syncSend(destination, message); if(sendResult.getSendStatus() != SendStatus.SEND_OK){ throw new RuntimeException("【智能巡检】事务提交后才发送MQ的消息发送失败"+message); } } }); } else { SendResult sendResult = rocketMQTemplate.syncSend(destination, message); if(sendResult.getSendStatus() != SendStatus.SEND_OK){ throw new RuntimeException("【智能巡检】事务提交后才发送MQ的消息发送失败"+message); } } } }