炳 mongo消费

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

@@ -56,31 +56,39 @@ public abstract class AbstractMdbRedisReceiver implements StreamListener<String,
     }
 
     private boolean consumerMessageHandler(String msg) {
-        boolean aBoolean = false;
+        boolean aBoolean = true;
         try {
             GoblinQueueBizMongoDto queueBizMongoDto = JsonUtils.fromJson(msg, GoblinQueueBizMongoDto.class);
-            if (null == queueBizMongoDto) {
-                aBoolean = true;
-            } else {
+            if (null != queueBizMongoDto) {
                 String collectName = queueBizMongoDto.getCollect(), columnName = queueBizMongoDto.getColumn();
-                String prefix = queueBizMongoDto.getBizId(), bizId = queueBizMongoDto.getBizId();
+                String prefix = queueBizMongoDto.getPrefix(), bizId = queueBizMongoDto.getBizId();
 
+                Object o;
                 switch (queueBizMongoDto.getOpType()) {
                     case 1:// insert
-                        mongoTemplate.insert(redisUtil.get(prefix.concat(bizId)), collectName);
-                        aBoolean = true;
+                        o = redisUtil.get(prefix.concat(bizId));
+                        if (null != o) {
+                            mongoTemplate.insert(o, collectName);
+                        }
                         break;
                     case 2:// update
-                        BasicDBObject basicDBObject = (BasicDBObject) AbstractMdbRedisReceiver.BASIC_DB_OBJECT.clone();
-                        UpdateResult updateResult = mongoTemplate.getCollection(collectName).updateOne(
-                                Query.query(Criteria.where(columnName).is(bizId)).getQueryObject(),
-                                basicDBObject.append("$set", mongoTemplate.getConverter().convertToMongoType(redisUtil.get(prefix.concat(bizId))))
-                        );
-                        aBoolean = updateResult.getModifiedCount() > 0;
+                        o = redisUtil.get(prefix.concat(bizId));
+                        if (null != o) {
+                            BasicDBObject basicDBObject = (BasicDBObject) AbstractMdbRedisReceiver.BASIC_DB_OBJECT.clone();
+                            UpdateResult updateResult = mongoTemplate.getCollection(collectName).updateOne(
+                                    Query.query(Criteria.where(columnName).is(bizId)).getQueryObject(),
+                                    basicDBObject.append("$set", mongoTemplate.getConverter().convertToMongoType(o))
+                            );
+                        }
+                        break;
+                    default:
+                        log.error("CONSUMER MSG ERR_HANDLE[未知的操作类型,{}]", msg);
+                        break;
                 }
             }
         } catch (Exception e) {
             log.error("CONSUMER MSG EX_HANDLE ==> [{}]:{}", this.getRedisStreamKey(), msg, e);
+            aBoolean = false;
         } finally {
             if (!aBoolean) {
                 HashMap<String, String> map = CollectionUtil.mapStringString();