提交 消费者

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

@@ -8,4 +8,9 @@ public class MQConst {
 
 
     public static final String QUEUES_SQL_MAIN = "queue.sql.main";
+
+
+    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";
 }

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

@@ -4,6 +4,7 @@ import com.liquidnet.common.mq.constant.MQConst;
 import com.liquidnet.commons.lang.util.JsonUtils;
 import com.liquidnet.service.base.SqlMapping;
 import com.liquidnet.service.consumer.service.IBaseDao;
+import com.rabbitmq.client.Channel;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.amqp.rabbit.annotation.RabbitListener;
 import org.springframework.stereotype.Component;
@@ -45,4 +46,23 @@ public class ConsumerProcessor {
             log.error("error:sqlMessage:{}", JsonUtils.toJson(sqlMessage), e);
         }
     }
+
+    @RabbitListener(queues = MQConst.QUEUES_SQL_ORDER_CREATE)
+    public void consumerOrderCreate(String msg, Channel channel) {
+        SqlMapping.SqlMessage sqlMessage = JsonUtils.fromJson(msg, 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);
+            } else {
+                channel.basicReject(100200300L, true);
+            }
+            log.debug("batchSqls result of execution:{}", rstBatchSqls);
+        } catch (Exception e) {
+            log.error("error:sqlMessage:{}", JsonUtils.toJson(sqlMessage), e);
+        }
+    }
 }