MQ的可靠性投递优化

call-back-service/src/main/java/cn/freemud/amp/service/OrderCallBackMQService.java

@@ -62,7 +62,8 @@ public class OrderCallBackMQService {
     private AssortmentOpenPlatformWxappManager openPlatformWxappManager;
     @Value("${saas.order.invoice.report.partnerIds}")
     private String invoicePartnerIds;
-
+    @Autowired
+    private SendMQConfirmCallback sendMQConfirmCallback;
 
     private static final String backOrdersChangeOrderStatusConsumerQueue = "program.backorders_change_order_status_consumer_queue";
     private static final String backOrdersNotifyActivityExchange = "program.backorders_notify_activity_exchange";
@@ -376,7 +377,6 @@ public class OrderCallBackMQService {
     }
 
 
-
     /**
      * 订单状态公共处理中心 queue
      *
@@ -392,7 +392,7 @@ public class OrderCallBackMQService {
                 Header header = new Header(MQAction.UPDATE.getAction(), "call-back-service-api",
                         OrderCallBackConfig.ORDER_STATUS_PROCESS_CENTER_QUEUE_ROUTING_KEY, OrderCallBackConfig.ORDER_STATUS_PROCESS_CENTER_QUEUE);
                 MQMessage<OrderCallBackRequestVo> message = new MQMessage<>(header, body);
-                mqService.convertAndSend(OrderCallBackConfig.TOPIC_EXCHANGE_NAME, OrderCallBackConfig.ORDER_STATUS_PROCESS_CENTER_QUEUE_ROUTING_KEY, message);
+                sendMQConfirmCallback.convertAndSend(OrderCallBackConfig.TOPIC_EXCHANGE_NAME, OrderCallBackConfig.ORDER_STATUS_PROCESS_CENTER_QUEUE_ROUTING_KEY, message);
                 ApiLog.infoMessage("send order_status_process_center_queue end orderCode:{}", body.getOrderCode());
             }
         } catch (Exception e) {

call-back-service/src/main/java/cn/freemud/amp/service/SendMQConfirmCallback.java

 package cn.freemud.amp.service;
 
+import cn.freemud.amqp.FastJsonMessageConverter;
 import lombok.extern.log4j.Log4j2;
 import org.springframework.amqp.core.Message;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.connection.CorrelationData;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Configurable;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Primary;
+import org.springframework.context.annotation.Scope;
 
 import javax.annotation.PostConstruct;
 
@@ -19,10 +24,24 @@ import javax.annotation.PostConstruct;
 @Log4j2
 public class SendMQConfirmCallback {
 
+    @Bean
+    @Scope("prototype")
+    @Primary
+    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
+        RabbitTemplate template = new RabbitTemplate(connectionFactory);
+        template.setMessageConverter(new FastJsonMessageConverter(Object.class));
+        template.afterPropertiesSet();
+        return template;
+    }
 
     @Autowired
     private RabbitTemplate rabbitTemplate;
 
+
+    public void convertAndSend(String exchange, String queueName, final Object object) {
+        this.rabbitTemplate.convertAndSend(exchange, queueName, object);
+    }
+
     /**
      * 定制 RabbitTemplate
      * <p>