~queue+log;

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

@@ -20,6 +20,7 @@ import org.springframework.data.mongodb.core.MongoTemplate;
 import org.springframework.data.mongodb.core.query.Criteria;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.data.mongodb.core.query.Update;
+import org.springframework.data.redis.connection.stream.RecordId;
 import org.springframework.data.redis.connection.stream.StreamRecords;
 import org.springframework.data.redis.core.StreamOperations;
 import org.springframework.stereotype.Component;
@@ -147,15 +148,17 @@ public class ConsumerGoblinBizArtworkGenRdsReceiver extends AbstractBizRedisRece
         toMqSqls.add(SQL_INSERT_ARTWORK);
         LinkedList<Object[]> initArtworkObjs = CollectionUtil.linkedListObjectArr();
         initArtworkObjs.add(new Object[]{digitalArtworkVo.getArtworkId(), skuId, uid, orderId, source, digitalArtworkVo.getState(), now});
-        sqlArtworkMap.put("message", SqlMapping.gets(toMqSqls, initArtworkObjs));
+        sqlArtworkMap.put(MQConst.QUEUE_MESSAGE_KEY, SqlMapping.gets(toMqSqls, initArtworkObjs));
         streamOperations.add(StreamRecords.mapBacked(sqlArtworkMap).withStreamKey(MQConst.GoblinQueue.SQL_ARTWORK_GEN.getKey()));
         // NFT发行购买
         HashMap<String, String> bizNftBuyMap = CollectionUtil.mapStringString();
         ObjectNode bizNftBuyNode = JsonUtils.OM().createObjectNode();
         bizNftBuyNode.put("nftOrderPayId", orderId).put("routerType", "zxinchain")
                 .put("skuId", skuId).put("userId", uid).put("buyTimestamp", DateUtil.Formatter.yyyyMMddHHmmss.format(now));
-        bizNftBuyMap.put("message", bizNftBuyNode.toString());
-        streamOperations.add(StreamRecords.mapBacked(bizNftBuyMap).withStreamKey(MQConst.GalaxyQueue.JSON_NFT_PUBLISH_AND_BUY.getKey()));
+        bizNftBuyMap.put(MQConst.QUEUE_MESSAGE_KEY, bizNftBuyNode.toString());
+        RecordId add = streamOperations.add(StreamRecords.mapBacked(bizNftBuyMap).withStreamKey(MQConst.GalaxyQueue.JSON_NFT_PUBLISH_AND_BUY.getKey()));
+        log.info("CONSUMER MSG GEN_ART_TQ[{},message={}]", JsonUtils.toJson(add), bizNftBuyMap);
+
         return true;
     }