修改 controller

liquidnet-bus-common/liquidnet-common-mq/src/main/java/com/liquidnet/common/mq/constant/MQConst.java

@@ -13,4 +13,8 @@ public class MQConst {
     public static final String EXCHANGES_LIQUIDNET_SQL_ORDER_CREADE = "liquidnet.sql.order.create";
     public static final String ROUTING_KEY_SQL_ORDER_CREATE = "order.create";
     public static final String QUEUES_SQL_ORDER_CREATE = "queue.sql.order.create";
+
+    public static final String EXCHANGES_LIQUIDNET_SQL_PERFORMANCE_LACK = "liquidnet.sql.performance.lack";
+    public static final String ROUTING_KEY_SQL_PERFORMANCE_LACK = "performance.lack";
+    public static final String QUEUES_SQL_PERFORMANCE_LACK = "queue.sql.performance.lack";
 }

liquidnet-bus-service/liquidnet-service-consumer/src/main/java/com/liquidnet/service/consumer/service/processor/ConsumerProcessor.java

@@ -8,6 +8,7 @@ import com.rabbitmq.client.Channel;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.amqp.rabbit.annotation.RabbitListener;
 import org.springframework.stereotype.Component;
+import org.springframework.amqp.core.Message;
 
 import javax.annotation.Resource;
 
@@ -48,17 +49,38 @@ public class ConsumerProcessor {
     }
 
     @RabbitListener(queues = MQConst.QUEUES_SQL_ORDER_CREATE)
-    public void consumerOrderCreate(String msg, Channel channel) {
-        SqlMapping.SqlMessage sqlMessage = JsonUtils.fromJson(msg, SqlMapping.SqlMessage.class);
+    public void consumerOrderCreate(Message msg, Channel channel) {
+        log.info("DEMO MESSAGE_GETBODY ==> Preparing:{}", new String(msg.getBody()));
+        SqlMapping.SqlMessage sqlMessage = JsonUtils.fromJson(new String(msg.getBody()), SqlMapping.SqlMessage.class);
+        log.debug("consumer sql_order_create ==> Preparing:{}", JsonUtils.toJson(sqlMessage.getSqls()));
+        log.debug("consumer sql_order_create ==> Parameters:{}", JsonUtils.toJson(sqlMessage.getArgs()));
+
+        try {
+            Boolean rstBatchSqls = baseDao.batchSqls(sqlMessage.getSqls(), sqlMessage.getArgs());
+            if (rstBatchSqls) {
+                channel.basicAck(msg.getMessageProperties().getDeliveryTag(), true);
+            } else {
+                channel.basicReject(msg.getMessageProperties().getDeliveryTag(), true);
+            }
+            log.debug("batchSqls result of execution:{}", rstBatchSqls);
+        } catch (Exception e) {
+            log.error("error:sqlMessage:{}", JsonUtils.toJson(sqlMessage), e);
+        }
+    }
+
+    @RabbitListener(queues = MQConst.QUEUES_SQL_PERFORMANCE_LACK)
+    public void consumerPerformanceLAck(Message msg, Channel channel) {
+        log.info("DEMO MESSAGE_GETBODY ==> Preparing:{}", new String(msg.getBody()));
+        SqlMapping.SqlMessage sqlMessage = JsonUtils.fromJson(new String(msg.getBody()), SqlMapping.SqlMessage.class);
         log.debug("consumer sql_order_create ==> Preparing:{}", JsonUtils.toJson(sqlMessage.getSqls()));
         log.debug("consumer sql_order_create ==> Parameters:{}", JsonUtils.toJson(sqlMessage.getArgs()));
 
         try {
             Boolean rstBatchSqls = baseDao.batchSqls(sqlMessage.getSqls(), sqlMessage.getArgs());
             if (rstBatchSqls) {
-                channel.basicAck(100200300L, true);
+                channel.basicAck(msg.getMessageProperties().getDeliveryTag(), true);
             } else {
-                channel.basicReject(100200300L, true);
+                channel.basicReject(msg.getMessageProperties().getDeliveryTag(), true);
             }
             log.debug("batchSqls result of execution:{}", rstBatchSqls);
         } catch (Exception e) {