fix:mq;

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

 package com.liquidnet.common.mq.constant;
 
 public class MQConst {
-    public static final String EXCHANGES_LIQUIDNET_SQL = "liquidnet.sql";
+//    public static final String EXCHANGES_LIQUIDNET_SQL = "liquidnet.sql";
+//    public static final String ROUTING_KEY_SQL = "sql";
+//    public static final String QUEUES_SQL_MAIN = "queue.sql.main";
 
-    public static final String ROUTING_KEY_SQL = "sql";
-    public static final String QUEUES_SQL_MAIN = "queue.sql.main";
+    /* -------------------------------------------------------- */
 
-    public static final String ROUTING_KEY_SQL_KYLIN_STATION = "sql.kylin.station";
-    public static final String QUEUES_SQL_KYLIN_STATION = "queue.sql.kylin.station";
+    // 用户中心
+    public static final String EX_LNS_SQL_UCENTER = "lns.sql.ucenter";
+
+    public static final String RK_SQL_UCENTER = "sql.ucenter";
+    public static final String QUEUES_SQL_UCENTER = "queue.sql.ucenter";
+    // 用户注册
+    public static final String RK_SQL_UREGISTER = "sql.uregister";
+    public static final String QUEUES_SQL_UREGISTER = "queue.sql.uregister";
+    // 会员购买
+    public static final String RK_SQL_UMEMBER = "sql.umember";
+    public static final String QUEUES_SQL_UMEMBER = "queue.sql.umember";
+
+    /* -------------------------------------------------------- */
 
     //缺票登记
     public static final String EXCHANGES_LIQUIDNET_SQL_PERFORMANCE_LACK = "liquidnet.sql.performance.lack";
@@ -33,4 +45,13 @@ public class MQConst {
     public static final String EXCHANGES_LIQUIDNET_SQL_ORDER_PAY = "liquidnet.sql.order.pay";
     public static final String ROUTING_KEY_SQL_ORDER_PAY = "order.pay";
     public static final String QUEUES_SQL_ORDER_PAY = "queue.sql.order.pay";
+
+    /* -------------------------------------------------------- */
+
+    // 验票更新
+    public static final String EX_LNS_SQL_STATION = "lns.sql.station";
+    public static final String RK_SQL_STATION = "sql.station";
+    public static final String QUEUES_SQL_STATION = "queue.sql.station";
+
+    /* -------------------------------------------------------- */
 }

liquidnet-bus-service/liquidnet-service-adam/liquidnet-service-adam-impl/src/main/java/com/liquidnet/service/adam/service/impl/AdamAddressesServiceImpl.java

@@ -68,7 +68,7 @@ public class AdamAddressesServiceImpl implements IAdamAddressesService {
 
         adamRdmService.delAddressesVoByUid(currentUid);
 
-        rabbitTemplate.convertSendAndReceive(MQConst.EXCHANGES_LIQUIDNET_SQL, MQConst.ROUTING_KEY_SQL,
+        rabbitTemplate.convertSendAndReceive(MQConst.EX_LNS_SQL_UCENTER, MQConst.RK_SQL_UCENTER,
                 SqlMapping.get("adam_addresses.add",
                         vo.getAddressesId(), vo.getUid(), vo.getName(), vo.getPhone(), vo.getProvince(), vo.getCity(), vo.getCounty(), vo.getAddress(), vo.getIsDefault(), vo.getState(), now
                 )
@@ -118,7 +118,7 @@ public class AdamAddressesServiceImpl implements IAdamAddressesService {
         }
         if (!CollectionUtils.isEmpty(toMqObjs)) {
             adamRdmService.setAddressesVoByUid(uid, vos);
-            rabbitTemplate.convertSendAndReceive(MQConst.EXCHANGES_LIQUIDNET_SQL, MQConst.ROUTING_KEY_SQL,
+            rabbitTemplate.convertSendAndReceive(MQConst.EX_LNS_SQL_UCENTER, MQConst.RK_SQL_UCENTER,
                     SqlMapping.get("adam_addresses.update.is_default", toMqObjs));
         }
     }
@@ -142,7 +142,7 @@ public class AdamAddressesServiceImpl implements IAdamAddressesService {
             List<AdamAddressesVo> vos = adamRdmService.getAddressesVoByUid(currentUid);
             adamRdmService.setAddressesVoByUid(updateVo.getUid(), this.collectionProcess(vos, parameter.getAddressesId(), updateVo));
 
-            rabbitTemplate.convertSendAndReceive(MQConst.EXCHANGES_LIQUIDNET_SQL, MQConst.ROUTING_KEY_SQL,
+            rabbitTemplate.convertSendAndReceive(MQConst.EX_LNS_SQL_UCENTER, MQConst.RK_SQL_UCENTER,
                     SqlMapping.get("adam_addresses.edit",
                             updateVo.getName(), updateVo.getPhone(), updateVo.getProvince(), updateVo.getCity(), updateVo.getCounty(), updateVo.getAddress(), now, updateVo.getAddressesId()
                     )
@@ -171,7 +171,7 @@ public class AdamAddressesServiceImpl implements IAdamAddressesService {
             vos.removeIf(r -> r.getAddressesId().equals(addressesId));
             adamRdmService.setAddressesVoByUid(uid, vos);
 
-            rabbitTemplate.convertSendAndReceive(MQConst.EXCHANGES_LIQUIDNET_SQL, MQConst.ROUTING_KEY_SQL,
+            rabbitTemplate.convertSendAndReceive(MQConst.EX_LNS_SQL_UCENTER, MQConst.RK_SQL_UCENTER,
                     SqlMapping.get("adam_addresses.remove", now, now, addressesId)
             );
         }

liquidnet-bus-service/liquidnet-service-adam/liquidnet-service-adam-impl/src/main/java/com/liquidnet/service/adam/service/impl/AdamCollectionServiceImpl.java

@@ -60,7 +60,7 @@ public class AdamCollectionServiceImpl implements IAdamCollectionService {
 
         mongoTemplate.insert(vo, AdamCollectBaseVo.class.getSimpleName());
 
-        rabbitTemplate.convertSendAndReceive(MQConst.EXCHANGES_LIQUIDNET_SQL, MQConst.ROUTING_KEY_SQL,
+        rabbitTemplate.convertSendAndReceive(MQConst.EX_LNS_SQL_UCENTER, MQConst.RK_SQL_UCENTER,
                 SqlMapping.get("adam_collection.add",
                         vo.getUid(), vo.getContentId(), vo.getType(), vo.getState(), now
                 )
@@ -91,7 +91,7 @@ public class AdamCollectionServiceImpl implements IAdamCollectionService {
             for (String c : contentIds) {
                 toMqObjs.add(new Object[]{now, uid, c});
             }
-            rabbitTemplate.convertSendAndReceive(MQConst.EXCHANGES_LIQUIDNET_SQL, MQConst.ROUTING_KEY_SQL,
+            rabbitTemplate.convertSendAndReceive(MQConst.EX_LNS_SQL_UCENTER, MQConst.RK_SQL_UCENTER,
                     SqlMapping.get("adam_collection.del", toMqObjs));
         }
     }

liquidnet-bus-service/liquidnet-service-adam/liquidnet-service-adam-impl/src/main/java/com/liquidnet/service/adam/service/impl/AdamDisposedServiceImpl.java

@@ -63,7 +63,7 @@ public class AdamDisposedServiceImpl implements IAdamDisposedService {
 
         mongoTemplate.insert(vo, AdamDisposedBaseVo.class.getSimpleName());
 
-        rabbitTemplate.convertSendAndReceive(MQConst.EXCHANGES_LIQUIDNET_SQL, MQConst.ROUTING_KEY_SQL,
+        rabbitTemplate.convertSendAndReceive(MQConst.EX_LNS_SQL_UCENTER, MQConst.RK_SQL_UCENTER,
                 SqlMapping.get("adam_disposed.add",
                         vo.getUid(), vo.getContentId(), vo.getType(), vo.getState(), now
                 )
@@ -95,7 +95,7 @@ public class AdamDisposedServiceImpl implements IAdamDisposedService {
             for (String c : contentIds) {
                 toMqObjs.add(new Object[]{now, uid, c});
             }
-            rabbitTemplate.convertSendAndReceive(MQConst.EXCHANGES_LIQUIDNET_SQL, MQConst.ROUTING_KEY_SQL,
+            rabbitTemplate.convertSendAndReceive(MQConst.EX_LNS_SQL_UCENTER, MQConst.RK_SQL_UCENTER,
                     SqlMapping.get("adam_disposed.del", toMqObjs));
         }
     }

liquidnet-bus-service/liquidnet-service-adam/liquidnet-service-adam-impl/src/main/java/com/liquidnet/service/adam/service/impl/AdamEntersServiceImpl.java

@@ -71,7 +71,7 @@ public class AdamEntersServiceImpl implements IAdamEntersService {
 
         adamRdmService.delEntersVoByUid(currentUid);
 
-        rabbitTemplate.convertSendAndReceive(MQConst.EXCHANGES_LIQUIDNET_SQL, MQConst.ROUTING_KEY_SQL,
+        rabbitTemplate.convertSendAndReceive(MQConst.EX_LNS_SQL_UCENTER, MQConst.RK_SQL_UCENTER,
                 SqlMapping.get("adam_enters.add",
                         vo.getEntersId(), vo.getUid(), vo.getType(), vo.getName(), vo.getMobile(), vo.getIdCard(), vo.getIsDefault(), vo.getState(), now
                 )
@@ -123,7 +123,7 @@ public class AdamEntersServiceImpl implements IAdamEntersService {
         }
         if (!CollectionUtils.isEmpty(toMqObjs)) {
             adamRdmService.setEntersVoByUid(uid, vos);
-            rabbitTemplate.convertSendAndReceive(MQConst.EXCHANGES_LIQUIDNET_SQL, MQConst.ROUTING_KEY_SQL,
+            rabbitTemplate.convertSendAndReceive(MQConst.EX_LNS_SQL_UCENTER, MQConst.RK_SQL_UCENTER,
                     SqlMapping.get("adam_enters.update.is_default", toMqObjs));
         }
     }
@@ -147,7 +147,7 @@ public class AdamEntersServiceImpl implements IAdamEntersService {
             List<AdamEntersVo> vos = adamRdmService.getEntersVoByUid(currentUid);
             adamRdmService.setEntersVoByUid(currentUid, this.collectionProcess(vos, parameter.getEntersId(), updateVo));
 
-            rabbitTemplate.convertSendAndReceive(MQConst.EXCHANGES_LIQUIDNET_SQL, MQConst.ROUTING_KEY_SQL,
+            rabbitTemplate.convertSendAndReceive(MQConst.EX_LNS_SQL_UCENTER, MQConst.RK_SQL_UCENTER,
                     SqlMapping.get("adam_enters.edit",
                             updateVo.getType(), updateVo.getName(), updateVo.getMobile(), updateVo.getIdCard(), updateVo.getIsDefault(), updateVo.getState(), now, updateVo.getEntersId()
                     )
@@ -176,7 +176,7 @@ public class AdamEntersServiceImpl implements IAdamEntersService {
             vos.removeIf(r -> r.getEntersId().equals(entersId));
             adamRdmService.setEntersVoByUid(uid, vos);
 
-            rabbitTemplate.convertSendAndReceive(MQConst.EXCHANGES_LIQUIDNET_SQL, MQConst.ROUTING_KEY_SQL,
+            rabbitTemplate.convertSendAndReceive(MQConst.EX_LNS_SQL_UCENTER, MQConst.RK_SQL_UCENTER,
                     SqlMapping.get("adam_enters.remove", now, now, entersId)
             );
         }

liquidnet-bus-service/liquidnet-service-adam/liquidnet-service-adam-impl/src/main/java/com/liquidnet/service/adam/service/impl/AdamMemberOrderServiceImpl.java

@@ -284,7 +284,7 @@ public class AdamMemberOrderServiceImpl extends ServiceImpl<AdamMemberOrderMappe
 
         adamRdmService.setShotMemberOrderVoByOrderNo(handleMemberOrderVo.getUid(), handleMemberOrderVo.getOrderNo(), handleMemberOrderVo);
 
-        rabbitTemplate.convertSendAndReceive(MQConst.EXCHANGES_LIQUIDNET_SQL, MQConst.ROUTING_KEY_SQL,
+        rabbitTemplate.convertSendAndReceive(MQConst.EX_LNS_SQL_UCENTER, MQConst.RK_SQL_UMEMBER,
                 SqlMapping.gets(toMqSqls, operationObjs, updateMemberOrderObjs));
         return ResponseDto.success();
     }
@@ -386,7 +386,7 @@ public class AdamMemberOrderServiceImpl extends ServiceImpl<AdamMemberOrderMappe
         adamRdmService.setUserMemberVoByUid(initMemberOrderVo.getUid(), initUserMemberVo);
         adamRdmService.delMemberCodeVoByCode(memberCodeVo.getCode());
 
-        rabbitTemplate.convertSendAndReceive(MQConst.EXCHANGES_LIQUIDNET_SQL, MQConst.ROUTING_KEY_SQL,
+        rabbitTemplate.convertSendAndReceive(MQConst.EX_LNS_SQL_UCENTER, MQConst.RK_SQL_UMEMBER,
                 SqlMapping.gets(toMqSqls, updateMemberCodeObjs, initMemberOrderObjs, initUserMemberObjs));
 
         AdamMemberOrderResult result = AdamMemberOrderResult.getNew();

liquidnet-bus-service/liquidnet-service-adam/liquidnet-service-adam-impl/src/main/java/com/liquidnet/service/adam/service/impl/AdamRealNameServiceImpl.java

@@ -49,7 +49,7 @@ public class AdamRealNameServiceImpl extends ServiceImpl<AdamRealNameMapper, Ada
                 realName.getState(),
                 realName.getCreatedAt()
         );
-        rabbitTemplate.convertSendAndReceive(MQConst.EXCHANGES_LIQUIDNET_SQL, MQConst.ROUTING_KEY_SQL,
+        rabbitTemplate.convertSendAndReceive(MQConst.EX_LNS_SQL_UCENTER, MQConst.RK_SQL_UCENTER,
                 SqlMapping.get("adam_real_name.add", paramList.toArray()));
     }
 }

liquidnet-bus-service/liquidnet-service-adam/liquidnet-service-adam-impl/src/main/java/com/liquidnet/service/adam/service/impl/AdamUserInfoServiceImpl.java

@@ -67,7 +67,7 @@ public class AdamUserInfoServiceImpl extends ServiceImpl<AdamUserInfoMapper, Ada
                 userInfoVo.getTagMe(),
                 userInfoVo.getUid()
         );
-        rabbitTemplate.convertSendAndReceive(MQConst.EXCHANGES_LIQUIDNET_SQL, MQConst.ROUTING_KEY_SQL,
+        rabbitTemplate.convertSendAndReceive(MQConst.EX_LNS_SQL_UCENTER, MQConst.RK_SQL_UCENTER,
                 SqlMapping.get(userInfoVo.getIsComplete() == 0 ? "adam_user_info.add" : "adam_user_info.edit", paramList.toArray()));
 
         adamRdmService.setUserInfoVoByUid(userInfoVo.getUid(), userInfoVo);
@@ -86,7 +86,7 @@ public class AdamUserInfoServiceImpl extends ServiceImpl<AdamUserInfoMapper, Ada
                 new FindOneAndUpdateOptions().returnDocument(ReturnDocument.BEFORE)
         );
         AdamUserInfoVo beforeUserInfoVo = BsonUtil.toBean(doc, AdamUserInfoVo.class);
-        rabbitTemplate.convertSendAndReceive(MQConst.EXCHANGES_LIQUIDNET_SQL, MQConst.ROUTING_KEY_SQL,
+        rabbitTemplate.convertSendAndReceive(MQConst.EX_LNS_SQL_UCENTER, MQConst.RK_SQL_UCENTER,
                 SqlMapping.get("adam_user.edit.mobile", mobile, now, uid));
 
         adamRdmService.delUidByMobile(beforeUserInfoVo.getMobile());

liquidnet-bus-service/liquidnet-service-adam/liquidnet-service-adam-impl/src/main/java/com/liquidnet/service/adam/service/impl/AdamUserServiceImpl.java

@@ -81,7 +81,7 @@ public class AdamUserServiceImpl implements IAdamUserService {
 
         mongoTemplate.insert(userInfoVo, AdamUserInfoVo.class.getSimpleName());
 
-        rabbitTemplate.convertSendAndReceive(MQConst.EXCHANGES_LIQUIDNET_SQL, MQConst.ROUTING_KEY_SQL,
+        rabbitTemplate.convertSendAndReceive(MQConst.EX_LNS_SQL_UCENTER, MQConst.RK_SQL_UREGISTER,
                 SqlMapping.get("adam_user.add", userInfoVo.getUid(), userInfoVo.getMobile(), userInfoVo.getState(), now)
         );
 
@@ -130,7 +130,7 @@ public class AdamUserServiceImpl implements IAdamUserService {
 
         adamRdmService.setUidByPlatformOpenId(param.getPlatform(), param.getOpenId(), uid);
 
-        rabbitTemplate.convertSendAndReceive(MQConst.EXCHANGES_LIQUIDNET_SQL, MQConst.ROUTING_KEY_SQL,
+        rabbitTemplate.convertSendAndReceive(MQConst.EX_LNS_SQL_UCENTER, MQConst.RK_SQL_UREGISTER,
                 SqlMapping.gets(toMqSqls, initUserObjs, initThirdPartObjs)
         );
         return userInfoVo;
@@ -153,7 +153,7 @@ public class AdamUserServiceImpl implements IAdamUserService {
 
         adamRdmService.delThirdPartVoListByUid(uid);
 
-        rabbitTemplate.convertSendAndReceive(MQConst.EXCHANGES_LIQUIDNET_SQL, MQConst.ROUTING_KEY_SQL,
+        rabbitTemplate.convertSendAndReceive(MQConst.EX_LNS_SQL_UCENTER, MQConst.RK_SQL_UCENTER,
                 SqlMapping.get(
                         "adam_third_party.add",
                         thirdPartInfoVo.getUid(), thirdPartInfoVo.getOpenId(), thirdPartInfoVo.getAvatar(), thirdPartInfoVo.getNickname(), thirdPartInfoVo.getPlatform(), thirdPartInfoVo.getState(), thirdPartInfoVo.getCreatedAt()
@@ -181,7 +181,7 @@ public class AdamUserServiceImpl implements IAdamUserService {
 
         adamRdmService.delThirdPartVoListByUid(bindUid);
 
-        rabbitTemplate.convertSendAndReceive(MQConst.EXCHANGES_LIQUIDNET_SQL, MQConst.ROUTING_KEY_SQL,
+        rabbitTemplate.convertSendAndReceive(MQConst.EX_LNS_SQL_UCENTER, MQConst.RK_SQL_UCENTER,
                 SqlMapping.get(
                         "adam_third_party.add",
                         thirdPartInfoVo.getUid(), thirdPartInfoVo.getOpenId(), thirdPartInfoVo.getAvatar(), thirdPartInfoVo.getNickname(), thirdPartInfoVo.getPlatform(), thirdPartInfoVo.getState(), thirdPartInfoVo.getCreatedAt()
@@ -214,7 +214,7 @@ public class AdamUserServiceImpl implements IAdamUserService {
         }
         adamRdmService.delThirdPartVoListByUid(uid);
 
-        rabbitTemplate.convertSendAndReceive(MQConst.EXCHANGES_LIQUIDNET_SQL, MQConst.ROUTING_KEY_SQL,
+        rabbitTemplate.convertSendAndReceive(MQConst.EX_LNS_SQL_UCENTER, MQConst.RK_SQL_UCENTER,
                 SqlMapping.get("adam_third_party.unbind", now, uid, platform));
     }
 
@@ -280,7 +280,7 @@ public class AdamUserServiceImpl implements IAdamUserService {
             objsAddresses.add(new Object[]{now, now, uid});
         }
 
-        rabbitTemplate.convertSendAndReceive(MQConst.EXCHANGES_LIQUIDNET_SQL, MQConst.ROUTING_KEY_SQL,
+        rabbitTemplate.convertSendAndReceive(MQConst.EX_LNS_SQL_UCENTER, MQConst.RK_SQL_UCENTER,
                 SqlMapping.gets(toMqSqls, objsUser));
     }
 

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

@@ -28,13 +28,120 @@ public class ConsumerProcessor {
     @Resource
     IBaseDao baseDao;
 
+//    @RabbitListener(bindings = @QueueBinding(
+//            exchange = @Exchange(MQConst.EXCHANGES_LIQUIDNET_SQL), key = MQConst.ROUTING_KEY_SQL,
+//            value = @Queue(MQConst.QUEUES_SQL_MAIN)
+//    ))
+////    @RabbitListener(queues = MQConst.QUEUES_SQL_MAIN)
+//    public void consumerSql(Message msg, Channel channel) {
+//        SqlMapping.SqlMessage sqlMessage = JsonUtils.fromJson(new String(msg.getBody()), SqlMapping.SqlMessage.class);
+//        log.debug("consumer sql ==> Preparing:{}", JsonUtils.toJson(sqlMessage.getSqls()));
+//        log.debug("consumer sql ==> Parameters:{}", JsonUtils.toJson(sqlMessage.getArgs()));
+//
+//        try {
+//            Boolean rstBatchSqls = baseDao.batchSqls(sqlMessage.getSqls(), sqlMessage.getArgs());
+//            log.debug("consumer sql result of execution:{}", rstBatchSqls);
+//            if (rstBatchSqls) {
+//                channel.basicAck(msg.getMessageProperties().getDeliveryTag(), false);
+//            } else {
+//                channel.basicReject(msg.getMessageProperties().getDeliveryTag(), true);
+//            }
+//        } catch (Exception e) {
+//            log.error("error:consumer sql:{}", JsonUtils.toJson(sqlMessage), e);
+//            try {
+//                channel.basicReject(msg.getMessageProperties().getDeliveryTag(), true);
+//            } catch (IOException ioException) {
+//                log.error("error:consumer sql:basicReject.msg.tag:{}", msg.getMessageProperties().getDeliveryTag(), ioException);
+//            }
+//        }
+//    }
+
+    @RabbitListener(bindings = @QueueBinding(
+            exchange = @Exchange(MQConst.EX_LNS_SQL_UCENTER), key = MQConst.RK_SQL_UCENTER,
+            value = @Queue(MQConst.QUEUES_SQL_UCENTER)
+    ))
+    public void consumerSqlForURegister(Message msg, Channel channel) {
+        SqlMapping.SqlMessage sqlMessage = JsonUtils.fromJson(new String(msg.getBody()), SqlMapping.SqlMessage.class);
+        log.debug("consumer sql ==> Preparing:{}", JsonUtils.toJson(sqlMessage.getSqls()));
+        log.debug("consumer sql ==> Parameters:{}", JsonUtils.toJson(sqlMessage.getArgs()));
+
+        try {
+            Boolean rstBatchSqls = baseDao.batchSqls(sqlMessage.getSqls(), sqlMessage.getArgs());
+            log.debug("consumer sql result of execution:{}", rstBatchSqls);
+            if (rstBatchSqls) {
+                channel.basicAck(msg.getMessageProperties().getDeliveryTag(), false);
+            } else {
+                channel.basicReject(msg.getMessageProperties().getDeliveryTag(), true);
+            }
+        } catch (Exception e) {
+            log.error("error:consumer sql:{}", JsonUtils.toJson(sqlMessage), e);
+            try {
+                channel.basicReject(msg.getMessageProperties().getDeliveryTag(), true);
+            } catch (IOException ioException) {
+                log.error("error:consumer sql:basicReject.msg.tag:{}", msg.getMessageProperties().getDeliveryTag(), ioException);
+            }
+        }
+    }
+
+    @RabbitListener(bindings = @QueueBinding(
+            exchange = @Exchange(MQConst.EX_LNS_SQL_UCENTER), key = MQConst.RK_SQL_UREGISTER,
+            value = @Queue(MQConst.QUEUES_SQL_UREGISTER)
+    ))
+    public void consumerSqlForUCenter(Message msg, Channel channel) {
+        SqlMapping.SqlMessage sqlMessage = JsonUtils.fromJson(new String(msg.getBody()), SqlMapping.SqlMessage.class);
+        log.debug("consumer sql ==> Preparing:{}", JsonUtils.toJson(sqlMessage.getSqls()));
+        log.debug("consumer sql ==> Parameters:{}", JsonUtils.toJson(sqlMessage.getArgs()));
+
+        try {
+            Boolean rstBatchSqls = baseDao.batchSqls(sqlMessage.getSqls(), sqlMessage.getArgs());
+            log.debug("consumer sql result of execution:{}", rstBatchSqls);
+            if (rstBatchSqls) {
+                channel.basicAck(msg.getMessageProperties().getDeliveryTag(), false);
+            } else {
+                channel.basicReject(msg.getMessageProperties().getDeliveryTag(), true);
+            }
+        } catch (Exception e) {
+            log.error("error:consumer sql:{}", JsonUtils.toJson(sqlMessage), e);
+            try {
+                channel.basicReject(msg.getMessageProperties().getDeliveryTag(), true);
+            } catch (IOException ioException) {
+                log.error("error:consumer sql:basicReject.msg.tag:{}", msg.getMessageProperties().getDeliveryTag(), ioException);
+            }
+        }
+    }
+
+    @RabbitListener(bindings = @QueueBinding(
+            exchange = @Exchange(MQConst.EX_LNS_SQL_UCENTER), key = MQConst.RK_SQL_UMEMBER,
+            value = @Queue(MQConst.QUEUES_SQL_UMEMBER)
+    ))
+    public void consumerSqlForUMember(Message msg, Channel channel) {
+        SqlMapping.SqlMessage sqlMessage = JsonUtils.fromJson(new String(msg.getBody()), SqlMapping.SqlMessage.class);
+        log.debug("consumer sql ==> Preparing:{}", JsonUtils.toJson(sqlMessage.getSqls()));
+        log.debug("consumer sql ==> Parameters:{}", JsonUtils.toJson(sqlMessage.getArgs()));
+
+        try {
+            Boolean rstBatchSqls = baseDao.batchSqls(sqlMessage.getSqls(), sqlMessage.getArgs());
+            log.debug("consumer sql result of execution:{}", rstBatchSqls);
+            if (rstBatchSqls) {
+                channel.basicAck(msg.getMessageProperties().getDeliveryTag(), false);
+            } else {
+                channel.basicReject(msg.getMessageProperties().getDeliveryTag(), true);
+            }
+        } catch (Exception e) {
+            log.error("error:consumer sql:{}", JsonUtils.toJson(sqlMessage), e);
+            try {
+                channel.basicReject(msg.getMessageProperties().getDeliveryTag(), true);
+            } catch (IOException ioException) {
+                log.error("error:consumer sql:basicReject.msg.tag:{}", msg.getMessageProperties().getDeliveryTag(), ioException);
+            }
+        }
+    }
+
     @RabbitListener(bindings = @QueueBinding(
-            exchange = @Exchange(MQConst.EXCHANGES_LIQUIDNET_SQL),
-            key = MQConst.ROUTING_KEY_SQL,
-            value = @Queue(MQConst.QUEUES_SQL_MAIN)
+            exchange = @Exchange(MQConst.EX_LNS_SQL_STATION), key = MQConst.RK_SQL_STATION,
+            value = @Queue(MQConst.QUEUES_SQL_STATION)
     ))
-//    @RabbitListener(queues = MQConst.QUEUES_SQL_MAIN)
-    public void consumerSql(Message msg, Channel channel) {
+    public void consumerSqlForStation(Message msg, Channel channel) {
         SqlMapping.SqlMessage sqlMessage = JsonUtils.fromJson(new String(msg.getBody()), SqlMapping.SqlMessage.class);
         log.debug("consumer sql ==> Preparing:{}", JsonUtils.toJson(sqlMessage.getSqls()));
         log.debug("consumer sql ==> Parameters:{}", JsonUtils.toJson(sqlMessage.getArgs()));

liquidnet-bus-service/liquidnet-service-kylin/liquidnet-service-kylin-impl/src/main/java/com/liquidnet/service/kylin/service/impl/KylinOrderTicketEntitiesServiceImpl.java

@@ -85,7 +85,7 @@ public class KylinOrderTicketEntitiesServiceImpl extends ServiceImpl<KylinOrderT
         BulkWriteResult bulkWriteResult = mongoTemplate.getCollection(KylinOrderTicketEntitiesVo.class.getSimpleName()).bulkWrite(list);
         log.info("bulkWriteResult:{}", JsonUtils.toJson(bulkWriteResult));
 
-        rabbitTemplate.convertSendAndReceive(MQConst.EXCHANGES_LIQUIDNET_SQL, MQConst.ROUTING_KEY_SQL_KYLIN_STATION,
+        rabbitTemplate.convertSendAndReceive(MQConst.EX_LNS_SQL_STATION, MQConst.RK_SQL_STATION,
                 SqlMapping.get("kylin_order_ticket_entities.updateStatusByStation", paramsList));
     }
 }