修改 消费

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-adam/src/main/java/com/liquidnet/service/consumer/adam/receiver/AbstractSqlRedisReceiver.java

@@ -46,7 +46,11 @@ public abstract class AbstractSqlRedisReceiver implements StreamListener<String,
         Boolean aBoolean = false;
         try {
             SqlMapping.SqlMessage sqlMessage = JsonUtils.fromJson(msg, SqlMapping.SqlMessage.class);
-            aBoolean = baseDao.batchSqls(sqlMessage.getSqls(), sqlMessage.getArgs());
+            if (sqlMessage == null) {
+                aBoolean = true;
+            } else {
+                aBoolean = baseDao.batchSqls(sqlMessage.getSqls(), sqlMessage.getArgs());
+            }
         } catch (Exception e) {
             log.error("CONSUMER SQL FAIL ==> {}", e.getMessage(), e);
         } finally {

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/receiver/AbstractSqlRedisReceiver.java

 package com.liquidnet.service.consumer.kylin.receiver;
 
+import com.liquidnet.common.cache.redis.util.RedisUtil;
 import com.liquidnet.commons.lang.util.CollectionUtil;
 import com.liquidnet.commons.lang.util.JsonUtils;
 import com.liquidnet.service.base.SqlMapping;
@@ -19,6 +20,8 @@ public abstract class AbstractSqlRedisReceiver implements StreamListener<String,
     private IBaseDao baseDao;
     @Autowired
     StringRedisTemplate stringRedisTemplate;
+    @Autowired
+    private RedisUtil redisUtil;
 
     @Override
     public void onMessage(MapRecord<String, String, String> message) {
@@ -46,7 +49,11 @@ public abstract class AbstractSqlRedisReceiver implements StreamListener<String,
         Boolean aBoolean = false;
         try {
             SqlMapping.SqlMessage sqlMessage = JsonUtils.fromJson(msg, SqlMapping.SqlMessage.class);
-            aBoolean = baseDao.batchSqls(sqlMessage.getSqls(), sqlMessage.getArgs());
+            if (sqlMessage == null) {
+                aBoolean = true;
+            } else {
+                aBoolean = baseDao.batchSqls(sqlMessage.getSqls(), sqlMessage.getArgs());
+            }
         } catch (Exception e) {
             log.error("CONSUMER SQL FAIL ==> {}", e.getMessage(), e);
         } finally {