解决事物提交与消息发送顺序问题

最近在线上发现了一个问题，mq的监听时常会报消息不存在的异常，关键代码如下：

public void sendMessage(MessageData message) throws Exception {
if (message == null) return;

// 持久化消息 ①
String messageId = rpTransactionControlService.createTransactionControl(message.getMessageBody(), message.getMessageType(), message.getQueue(), delayTime, message.getField1(), message.getField2(), message.getField3());
log.info("create transaction control messageid = " + messageId);
message.setMessageId(messageId);

// 发送消息 ②
ActiveMQClientPool.getInstance().sendMsg(sendMessage, message.getQueue(), message.getDelayTime() * 1000);

}

导致的原因就是 ②已经消息发送了，但是①还没有事物提交，就导致了问题。

解决办法 1、 增加延迟发送 。

                2、 增加事物监听。

针对1方法，如果是activemq，有一个需要注意的地方， 需要修改activemq.xml  

<broker xmlns="http://activemq.apache.org/schema/core" brokerName="localhost" dataDirectory="${activemq.data}" schedulerSupport="true">

 即，增加 schedulerSupport="true" 参数

针对2方法， 需要先创建一个 TransactionalMessageListener 类

@Component
@Slf4j
public class TransactionalMessageListener {

@TransactionalEventListener(fallbackExecution = true, phase = TransactionPhase.AFTER_COMMIT)
public void afterCommit(EventMessage eventMessage) throws Exception {

if (eventMessage == null) return;

if (eventMessage.getEventType() == null) throw new NullPointerException("eventType");

if (eventMessage.getEventType() == EventType.ACTIVE_MQ) { // 防止事物已经还没有提交，mq监听器就已经收到了消息

if (eventMessage.getData() instanceof MessageData) {
MessageData message = (MessageData) eventMessage.getData();
String sendMessage = JSONObject.toJSONString(message);
log.info("push msg to activeMq, context :{ " + sendMessage + "}");
try {
ActiveMQClientPool.getInstance().sendMsg(sendMessage, message.getQueue(), message.getDelayTime() * 1000);
} catch (Exception e) {
log.error("push msg error to activeMq：{" + sendMessage + "}", e);
}
}

}
}

}

public class EventMessage<T> {

public EventMessage(EventType eventType, T data) {
this.eventType = eventType;
this.data = data;
}

private EventType eventType;

private T data;

public EventType getEventType() {
return eventType;
}

public void setEventType(EventType eventType) {
this.eventType = eventType;
}

public T getData() {
return data;
}

public void setData(T data) {
this.data = data;
}
}

然后修改原来逻辑如下

@Autowired
ApplicationEventPublisher publisher;

public void sendMessage(MessageData message) throws Exception {
if (message == null) return;

String messageId = rpTransactionControlService.createTransactionControl(message.getMessageBody(), message.getMessageType(), message.getQueue(), delayTime, message.getField1(), message.getField2(), message.getField3());
log.info("create transaction control messageid = " + messageId);
message.setMessageId(messageId);

        publisher.publishEvent(new EventMessage<MessageData>(EventType.MQ, message));

}

这里我需要对这个进行解释一下：   @TransactionalEventListener(fallbackExecution = true, phase = TransactionPhase.AFTER_COMMIT)     

fallbackExecution = true 是为了保证没有事物的时候也能正常收到消息

 phase = TransactionPhase.AFTER_COMMIT  代表提交后监听