最近在線上發現了一個問題，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 代表提交後監聽

解決事物提交與消息發送順序問題