feat: TxMsg 重试支持指定 connectionName

Author: cadecode

common/plugin/mq/src/main/java/com/github/cadecode/uniboot/common/plugin/mq/model/BaseTxMsg.java

@@ -20,4 +20,6 @@ public interface BaseTxMsg {
 
     String getMessage();
 
+    String getConnectionName();
+
 }

common/plugin/mq/src/main/java/com/github/cadecode/uniboot/common/plugin/mq/model/TxMsg.java

@@ -34,4 +34,7 @@ public class TxMsg implements BaseTxMsg {
     private String routingKey;
 
     private String message;
+
+    private String connectionName;
+
 }

framework/framework_base/src/main/java/com/github/cadecode/uniboot/framework/base/plugin/bean/po/PlgMqMsg.java

@@ -36,6 +36,8 @@ public class PlgMqMsg {
 
     private String message;
 
+    private String connectionName;
+
     private SendStateEnum sendState;
 
     private ConsumeStateEnum consumeState;

framework/framework_base/src/main/java/com/github/cadecode/uniboot/framework/base/plugin/bean/vo/PlgMqMsgVo.java

@@ -34,6 +34,7 @@ public static class PlgMqMsgPageResVo {
         private String bizKey;
         private String exchange;
         private String routingKey;
+        private String connectionName;
         private String message;
         private SendStateEnum sendState;
         private ConsumeStateEnum consumeState;

framework/framework_base/src/main/java/com/github/cadecode/uniboot/framework/base/plugin/handler/MqTxMsgHandler.java

@@ -25,6 +25,7 @@
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.ReturnedMessage;
 import org.springframework.amqp.rabbit.connection.CorrelationData;
+import org.springframework.amqp.rabbit.connection.SimpleResourceHolder;
 import org.springframework.stereotype.Component;
 
 import java.util.Date;
@@ -143,6 +144,11 @@ public void checkBeforeSend(BaseTxMsg txMsg, MsgOption msgOption) {
         if (ObjUtil.isEmpty(txMsg.getId())) {
             ((TxMsg) txMsg).setId(String.valueOf(IdWorker.getId(txMsg)));
         }
+        // 填充 connectionName
+        if (ObjUtil.isEmpty(txMsg.getConnectionName())) {
+            String connectionName = (String) SimpleResourceHolder.get(RabbitUtil.template().getConnectionFactory());
+            ((TxMsg) txMsg).setConnectionName(connectionName);
+        }
     }
 
     @Override
@@ -190,18 +196,27 @@ public void sendCommitted(BaseTxMsg txMsg, MsgOption msgOption) {
     }
 
     private void doSendCommittedOrRetry(BaseTxMsg txMsg) {
-        // 放入 txMsg id
-        CorrelationData correlationData = new CorrelationData(txMsg.getId());
-        RabbitUtil.send(txMsg.getExchange(), txMsg.getRoutingKey(), txMsg.getMessage(), msg -> {
-            // 设置事务消息 head 标记
-            msg.getMessageProperties().getHeaders().put(HEAD_TX_MSG_ID, txMsg.getId());
-            msg.getMessageProperties().getHeaders().put(HEAD_TX_MSG_BIZ_TYPE, txMsg.getBizType());
-            msg.getMessageProperties().getHeaders().put(HEAD_TX_MSG_BIZ_KEY, txMsg.getBizKey());
-            // 设置 ReturnedMessage，用于 callback 中获取消息
-            ReturnedMessage defaultReturnedMessage = new ReturnedMessage(msg, 0, DEFAULT_TX_MSG_REPLY_TEXT, null, null);
-            correlationData.setReturned(defaultReturnedMessage);
-            return msg;
-        }, correlationData);
+        try {
+            if (ObjUtil.isNotNull(txMsg.getConnectionName())) {
+                SimpleResourceHolder.bind(RabbitUtil.template().getConnectionFactory(), txMsg.getConnectionName());
+            }
+            // 放入 txMsg id
+            CorrelationData correlationData = new CorrelationData(txMsg.getId());
+            RabbitUtil.send(txMsg.getExchange(), txMsg.getRoutingKey(), txMsg.getMessage(), msg -> {
+                // 设置事务消息 head 标记
+                msg.getMessageProperties().getHeaders().put(HEAD_TX_MSG_ID, txMsg.getId());
+                msg.getMessageProperties().getHeaders().put(HEAD_TX_MSG_BIZ_TYPE, txMsg.getBizType());
+                msg.getMessageProperties().getHeaders().put(HEAD_TX_MSG_BIZ_KEY, txMsg.getBizKey());
+                // 设置 ReturnedMessage，用于 callback 中获取消息
+                ReturnedMessage defaultReturnedMessage = new ReturnedMessage(msg, 0, DEFAULT_TX_MSG_REPLY_TEXT, null, null);
+                correlationData.setReturned(defaultReturnedMessage);
+                return msg;
+            }, correlationData);
+        } finally {
+            if (ObjUtil.isNotNull(txMsg.getConnectionName())) {
+                SimpleResourceHolder.unbindIfPossible(RabbitUtil.template().getConnectionFactory());
+            }
+        }
     }
 
     @Override