合并nft消费端到kylin

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/pom.xml

@@ -58,6 +58,21 @@
             <artifactId>liquidnet-common-sms</artifactId>
             <version>1.0-SNAPSHOT</version>
         </dependency>
+        <dependency>
+            <groupId>com.liquidnet</groupId>
+            <artifactId>liquidnet-common-third-zxlnft</artifactId>
+            <version>1.0-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>com.liquidnet</groupId>
+            <artifactId>liquidnet-common-third-antchain</artifactId>
+            <version>1.0-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>com.liquidnet</groupId>
+            <artifactId>liquidnet-service-galaxy-api</artifactId>
+            <version>1.0-SNAPSHOT</version>
+        </dependency>
     </dependencies>
 
     <build>

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/config/ConsumerJsonNftPublishAndBuyConfig.java

+package com.liquidnet.service.consumer.kylin.config;
+
+import com.liquidnet.common.cache.redis.config.RedisStreamConfig;
+import com.liquidnet.service.base.constant.MQConst;
+import com.liquidnet.service.consumer.kylin.receiver.JsonRedisJsonNftPublishAndBuyReceiver;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.connection.stream.Consumer;
+import org.springframework.data.redis.connection.stream.MapRecord;
+import org.springframework.data.redis.connection.stream.ReadOffset;
+import org.springframework.data.redis.connection.stream.StreamOffset;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.stream.StreamMessageListenerContainer;
+import org.springframework.data.redis.stream.Subscription;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * @author AnJiabin <anjiabin@zhengzai.tv>
+ * @version V1.0
+ * @Description: 区块链nft订单相关
+ * @class: ConsumerSqlNftOrderInfoConfig
+ * @Package com.liquidnet.service.consumer.galaxy.config
+ * @Copyright: LightNet @ Copyright (c) 2022
+ * @date 2022/3/25 14:45
+ */
+@Configuration
+public class ConsumerJsonNftPublishAndBuyConfig extends RedisStreamConfig {
+    @Autowired
+    private JsonRedisJsonNftPublishAndBuyReceiver jsonNftPublishAndBuyReceiver;
+
+    @Autowired
+    StringRedisTemplate stringRedisTemplate;
+
+    @Bean
+    public List<Subscription> subscriptionJsonNftPublishAndBuy(RedisConnectionFactory factory) {
+        List<Subscription> subscriptionList = new ArrayList<>();
+        MQConst.GalaxyQueue stream = MQConst.GalaxyQueue.JSON_NFT_PUBLISH_AND_BUY;
+
+        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
+        for (int i = 0; i < 20; i++) {
+            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
+            subscriptionList.add(listenerContainer.receiveAutoAck(
+                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
+                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), jsonNftPublishAndBuyReceiver
+            ));
+            listenerContainer.start();
+        }
+        return subscriptionList;
+    }
+}

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/config/ConsumerSqlNftOrderInfoConfig.java

+package com.liquidnet.service.consumer.kylin.config;
+
+import com.liquidnet.common.cache.redis.config.RedisStreamConfig;
+import com.liquidnet.service.base.constant.MQConst;
+import com.liquidnet.service.consumer.galaxy.receiver.SqlRedisSqlSeriesInfoReceiver;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.connection.stream.Consumer;
+import org.springframework.data.redis.connection.stream.MapRecord;
+import org.springframework.data.redis.connection.stream.ReadOffset;
+import org.springframework.data.redis.connection.stream.StreamOffset;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.stream.StreamMessageListenerContainer;
+import org.springframework.data.redis.stream.Subscription;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * @author AnJiabin <anjiabin@zhengzai.tv>
+ * @version V1.0
+ * @Description: 区块链nft订单相关
+ * @class: ConsumerSqlNftOrderInfoConfig
+ * @Package com.liquidnet.service.consumer.galaxy.config
+ * @Copyright: LightNet @ Copyright (c) 2022
+ * @date 2022/3/25 14:45
+ */
+@Configuration
+public class ConsumerSqlNftOrderInfoConfig extends RedisStreamConfig {
+    @Autowired
+    private SqlRedisSqlSeriesInfoReceiver sqlSeriesInfoReceiver;
+
+    @Autowired
+    StringRedisTemplate stringRedisTemplate;
+
+    @Bean
+    public List<Subscription> subscriptionSqlNftOrderInfo(RedisConnectionFactory factory) {
+        List<Subscription> subscriptionList = new ArrayList<>();
+        MQConst.GalaxyQueue stream = MQConst.GalaxyQueue.SQL_NFT_ORDER_INFO;
+
+        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
+        for (int i = 0; i < 10; i++) {
+            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
+            subscriptionList.add(listenerContainer.receiveAutoAck(
+                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
+                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), sqlSeriesInfoReceiver
+            ));
+            listenerContainer.start();
+        }
+        return subscriptionList;
+    }
+}

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/config/ConsumerSqlNftTradeInfoConfig.java

+package com.liquidnet.service.consumer.kylin.config;
+
+import com.liquidnet.common.cache.redis.config.RedisStreamConfig;
+import com.liquidnet.service.base.constant.MQConst;
+import com.liquidnet.service.consumer.galaxy.receiver.SqlRedisSqlSeriesInfoReceiver;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.connection.stream.Consumer;
+import org.springframework.data.redis.connection.stream.MapRecord;
+import org.springframework.data.redis.connection.stream.ReadOffset;
+import org.springframework.data.redis.connection.stream.StreamOffset;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.stream.StreamMessageListenerContainer;
+import org.springframework.data.redis.stream.Subscription;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * @author AnJiabin <anjiabin@zhengzai.tv>
+ * @version V1.0
+ * @Description: 区块链nft交易相关
+ * @class: ConsumerSqlNftTradeInfoConfig
+ * @Package com.liquidnet.service.consumer.galaxy.config
+ * @Copyright: LightNet @ Copyright (c) 2022
+ * @date 2022/3/25 14:45
+ */
+@Configuration
+public class ConsumerSqlNftTradeInfoConfig extends RedisStreamConfig {
+    @Autowired
+    private SqlRedisSqlSeriesInfoReceiver sqlSeriesInfoReceiver;
+
+    @Autowired
+    StringRedisTemplate stringRedisTemplate;
+
+    @Bean
+    public List<Subscription> subscriptionSqlNftTradeInfo(RedisConnectionFactory factory) {
+        List<Subscription> subscriptionList = new ArrayList<>();
+        MQConst.GalaxyQueue stream = MQConst.GalaxyQueue.SQL_NFT_TRADE_INFO;
+
+        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
+        for (int i = 0; i < 10; i++) {
+            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
+            subscriptionList.add(listenerContainer.receiveAutoAck(
+                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
+                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), sqlSeriesInfoReceiver
+            ));
+            listenerContainer.start();
+        }
+        return subscriptionList;
+    }
+}

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/config/ConsumerSqlSeriesInfoConfig.java

+package com.liquidnet.service.consumer.kylin.config;
+
+import com.liquidnet.common.cache.redis.config.RedisStreamConfig;
+import com.liquidnet.service.base.constant.MQConst;
+import com.liquidnet.service.consumer.galaxy.receiver.SqlRedisSqlSeriesInfoReceiver;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.connection.stream.Consumer;
+import org.springframework.data.redis.connection.stream.MapRecord;
+import org.springframework.data.redis.connection.stream.ReadOffset;
+import org.springframework.data.redis.connection.stream.StreamOffset;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.stream.StreamMessageListenerContainer;
+import org.springframework.data.redis.stream.Subscription;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * @author AnJiabin <anjiabin@zhengzai.tv>
+ * @version V1.0
+ * @Description: 区块链系列信息
+ * @class: ConsumerSqlSeriesInfoConfig
+ * @Package com.liquidnet.service.consumer.galaxy.config
+ * @Copyright: LightNet @ Copyright (c) 2022
+ * @date 2022/3/25 14:45
+ */
+@Configuration
+public class ConsumerSqlSeriesInfoConfig extends RedisStreamConfig {
+    @Autowired
+    private SqlRedisSqlSeriesInfoReceiver sqlSeriesInfoReceiver;
+
+    @Autowired
+    StringRedisTemplate stringRedisTemplate;
+
+    @Bean
+    public List<Subscription> subscriptionSqlSeriesInfo(RedisConnectionFactory factory) {
+        List<Subscription> subscriptionList = new ArrayList<>();
+        MQConst.GalaxyQueue stream = MQConst.GalaxyQueue.SQL_SERIES_INFO;
+
+        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
+        for (int i = 0; i < 10; i++) {
+            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
+            subscriptionList.add(listenerContainer.receiveAutoAck(
+                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
+                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), sqlSeriesInfoReceiver
+            ));
+            listenerContainer.start();
+        }
+        return subscriptionList;
+    }
+}

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/config/ConsumerSqlUserInfoConfig.java

+package com.liquidnet.service.consumer.kylin.config;
+
+import com.liquidnet.common.cache.redis.config.RedisStreamConfig;
+import com.liquidnet.service.base.constant.MQConst;
+import com.liquidnet.service.consumer.galaxy.receiver.SqlRedisSqlUserInfoReceiver;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.connection.stream.Consumer;
+import org.springframework.data.redis.connection.stream.MapRecord;
+import org.springframework.data.redis.connection.stream.ReadOffset;
+import org.springframework.data.redis.connection.stream.StreamOffset;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.stream.StreamMessageListenerContainer;
+import org.springframework.data.redis.stream.Subscription;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * @author AnJiabin <anjiabin@zhengzai.tv>
+ * @version V1.0
+ * @Description: 区块链用户注册
+ * @class: ConsumerSqlUserInfoConfig
+ * @Package com.liquidnet.service.consumer.dragon.config
+ * @Copyright: LightNet @ Copyright (c) 2022
+ * @date 2022/3/22 14:45
+ */
+@Configuration
+public class ConsumerSqlUserInfoConfig extends RedisStreamConfig {
+    @Autowired
+    private SqlRedisSqlUserInfoReceiver sqlUserInfoReceiver;
+
+    @Autowired
+    StringRedisTemplate stringRedisTemplate;
+
+    @Bean
+    public List<Subscription> subscriptionSqlUserInfo(RedisConnectionFactory factory) {
+        List<Subscription> subscriptionList = new ArrayList<>();
+        MQConst.GalaxyQueue stream = MQConst.GalaxyQueue.SQL_USER_INFO;
+
+        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
+        for (int i = 0; i < 10; i++) {
+            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
+            subscriptionList.add(listenerContainer.receiveAutoAck(
+                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
+                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), sqlUserInfoReceiver
+            ));
+            listenerContainer.start();
+        }
+        return subscriptionList;
+    }
+}

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

+package com.liquidnet.service.consumer.kylin.receiver;
+
+import com.liquidnet.commons.lang.util.CollectionUtil;
+import com.liquidnet.commons.lang.util.JsonUtils;
+import com.liquidnet.service.consumer.galaxy.service.processor.ConsumerJsonNftPublishAndBuyProcessor;
+import com.liquidnet.service.galaxy.dto.param.GalaxyNftPublishAndBuyReqDto;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.connection.stream.MapRecord;
+import org.springframework.data.redis.connection.stream.StreamRecords;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.stream.StreamListener;
+
+import java.util.HashMap;
+
+@Slf4j
+public abstract class AbstractJsonRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> {
+    @Autowired
+    StringRedisTemplate stringRedisTemplate;
+
+    @Autowired
+    private ConsumerJsonNftPublishAndBuyProcessor jsonNftPublishAndBuyProcessor;
+
+    @Override
+    public void onMessage(MapRecord<String, String, String> message) {
+        String redisStreamKey = this.getRedisStreamKey();
+        log.debug("CONSUMER MSG[streamKey:{},messageId:{},stream:{},body:{}]", redisStreamKey, message.getId(), message.getStream(), message.getValue());
+        boolean result = this.consumerMessageHandler(message.getValue().get("message"));
+        log.info("CONSUMER MSG RESULT:{} ==> [{}]MESSAGE_ID:{}", result, redisStreamKey, message.getId());
+
+        try {
+            stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message);
+        } catch (Exception e) {
+            log.error("#CONSUMER MSG EX_ACK ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e);
+        }
+        try {
+            stringRedisTemplate.opsForStream().delete(redisStreamKey, message.getId());
+        } catch (Exception e) {
+            log.error("#CONSUMER MSG EX_DEL ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e);
+        }
+    }
+
+    private boolean consumerMessageHandler(String msg) {
+        boolean aBoolean = false;
+        try {
+            GalaxyNftPublishAndBuyReqDto textMessage = JsonUtils.fromJson(msg, GalaxyNftPublishAndBuyReqDto.class);
+            if (textMessage == null) {
+                aBoolean = true;
+            } else {
+                //执行计数
+                jsonNftPublishAndBuyProcessor.nftPublishAndBuy(textMessage);
+                aBoolean = true;
+            }
+        } catch (Exception e) {
+            log.error("CONSUMER MSG EX_HANDLE ==> [{}]:{}", this.getRedisStreamKey(), msg, e);
+        } finally {
+            if (!aBoolean) {
+                HashMap<String, String> map = CollectionUtil.mapStringString();
+                map.put("message", msg);
+                stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(this.getRedisStreamKey()));
+            }
+        }
+        return aBoolean;
+    }
+
+    protected abstract String getRedisStreamKey();
+
+    protected abstract String getRedisStreamGroup();
+}

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

@@ -4,7 +4,7 @@ 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.OrderCloseMapping;
-import com.liquidnet.service.consumer.kylin.Utils.KylinUtils;
+import com.liquidnet.service.consumer.kylin.utils.KylinUtils;
 import com.liquidnet.service.kylin.constant.KylinRedisConst;
 import com.liquidnet.service.kylin.dto.vo.mongo.KylinOrderTicketVo;
 import lombok.extern.slf4j.Slf4j;

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

@@ -4,7 +4,7 @@ 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.OrderCloseMapping;
-import com.liquidnet.service.consumer.kylin.Utils.KylinUtils;
+import com.liquidnet.service.consumer.kylin.utils.KylinUtils;
 import com.liquidnet.service.kylin.dto.vo.mongo.KylinOrderTicketEntitiesVo;
 import com.liquidnet.service.kylin.dto.vo.mongo.KylinOrderTicketVo;
 import lombok.extern.slf4j.Slf4j;

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

+package com.liquidnet.service.consumer.kylin.receiver;
+
+import com.liquidnet.service.base.constant.MQConst;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+public class JsonRedisJsonNftPublishAndBuyReceiver extends AbstractJsonRedisReceiver {
+
+    @Override
+    protected String getRedisStreamKey() {
+        return MQConst.GalaxyQueue.JSON_NFT_PUBLISH_AND_BUY.getKey();
+    }
+
+    @Override
+    protected String getRedisStreamGroup() {
+        return MQConst.GalaxyQueue.JSON_NFT_PUBLISH_AND_BUY.getGroup();
+    }
+}

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

+package com.liquidnet.service.consumer.kylin.receiver;
+
+import com.liquidnet.service.base.constant.MQConst;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+public class SqlRedisSqlNftOrderInfoReceiver extends AbstractSqlRedisReceiver {
+
+    @Override
+    protected String getRedisStreamKey() {
+        return MQConst.GalaxyQueue.SQL_NFT_ORDER_INFO.getKey();
+    }
+
+    @Override
+    protected String getRedisStreamGroup() {
+        return MQConst.GalaxyQueue.SQL_NFT_ORDER_INFO.getGroup();
+    }
+}

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

+package com.liquidnet.service.consumer.kylin.receiver;
+
+import com.liquidnet.service.base.constant.MQConst;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+public class SqlRedisSqlNftTradeInfoReceiver extends AbstractSqlRedisReceiver {
+
+    @Override
+    protected String getRedisStreamKey() {
+        return MQConst.GalaxyQueue.SQL_NFT_TRADE_INFO.getKey();
+    }
+
+    @Override
+    protected String getRedisStreamGroup() {
+        return MQConst.GalaxyQueue.SQL_NFT_TRADE_INFO.getGroup();
+    }
+}

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

+package com.liquidnet.service.consumer.kylin.receiver;
+
+import com.liquidnet.service.base.constant.MQConst;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+public class SqlRedisSqlSeriesInfoReceiver extends AbstractSqlRedisReceiver {
+
+    @Override
+    protected String getRedisStreamKey() {
+        return MQConst.GalaxyQueue.SQL_SERIES_INFO.getKey();
+    }
+
+    @Override
+    protected String getRedisStreamGroup() {
+        return MQConst.GalaxyQueue.SQL_SERIES_INFO.getGroup();
+    }
+}

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

+package com.liquidnet.service.consumer.kylin.receiver;
+
+import com.liquidnet.service.base.constant.MQConst;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+public class SqlRedisSqlUserInfoReceiver extends AbstractSqlRedisReceiver {
+
+    @Override
+    protected String getRedisStreamKey() {
+        return MQConst.GalaxyQueue.SQL_USER_INFO.getKey();
+    }
+
+    @Override
+    protected String getRedisStreamGroup() {
+        return MQConst.GalaxyQueue.SQL_USER_INFO.getGroup();
+    }
+}

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/utils/DataUtils.java

+package com.liquidnet.service.consumer.kylin.utils;
+
+import com.liquidnet.common.cache.redis.util.RedisUtil;
+import com.liquidnet.commons.lang.util.BeanUtil;
+import com.liquidnet.commons.lang.util.DateUtil;
+import com.liquidnet.commons.lang.util.JsonUtils;
+import com.liquidnet.service.base.SqlMapping;
+import com.liquidnet.service.base.constant.MQConst;
+import com.liquidnet.service.galaxy.constant.GalaxyConstant;
+import com.liquidnet.service.galaxy.dto.bo.GalaxyNftOrderBo;
+import com.liquidnet.service.galaxy.dto.bo.GalaxySeriesNftInfoBo;
+import com.liquidnet.service.galaxy.dto.bo.GalaxyUserInfoBo;
+import com.liquidnet.service.galaxy.dto.vo.mongo.GalaxyNftOrderVo;
+import com.liquidnet.service.galaxy.dto.vo.mongo.GalaxySeriesNftInfoVo;
+import com.liquidnet.service.galaxy.dto.vo.mongo.GalaxyUserInfoVo;
+import com.mongodb.client.result.UpdateResult;
+import lombok.extern.slf4j.Slf4j;
+import org.bson.Document;
+import org.springframework.beans.factory.annotation.Autowired;
+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.stereotype.Component;
+
+import java.time.LocalDateTime;
+
+@Slf4j
+@Component
+public class DataUtils {
+    @Autowired
+    private RedisUtil redisUtil;
+
+    @Autowired
+    private MongoTemplate mongoTemplate;
+
+    @Autowired
+    private QueueUtil queueUtil;
+
+    private long keyExpireTime = 3600*24*30;
+
+    public GalaxyUserInfoBo getGalaxyUserInfo(String routerType, String userId) {
+        Object obj = redisUtil.get(GalaxyConstant.REDIS_KEY_GALAXY_USER.concat(routerType).concat(":") + userId);
+        if(obj!=null){
+            return (GalaxyUserInfoBo) obj;
+        }else {
+            GalaxyUserInfoVo userInfoVo = mongoTemplate.findOne(Query.query(Criteria.where("userId").is(userId)), GalaxyUserInfoVo.class, GalaxyUserInfoVo.class.getSimpleName());
+            if (userInfoVo == null) {
+                return null;
+            }
+            GalaxyUserInfoBo userInfoBo = GalaxyUserInfoBo.getNew();
+            BeanUtil.copy(userInfoVo,userInfoBo);
+
+            redisUtil.set(GalaxyConstant.REDIS_KEY_GALAXY_USER.concat(routerType).concat(":") + userId, userInfoBo);
+            return userInfoBo;
+        }
+    }
+
+    public GalaxyNftOrderBo getNftOrderBo(String routerType, String nftOrderPayId) {
+        Object obj = redisUtil.get(GalaxyConstant.REDIS_KEY_GALAXY_TRADE.concat(routerType).concat(":") + nftOrderPayId);
+        if(obj!=null){
+            return (GalaxyNftOrderBo) obj;
+        }else {
+            GalaxyNftOrderVo nftOrderVo = mongoTemplate.findOne(Query.query(Criteria.where("nftOrderPayId").is(nftOrderPayId)), GalaxyNftOrderVo.class, GalaxyNftOrderVo.class.getSimpleName());
+            if (nftOrderVo == null) {
+                return null;
+            }
+            GalaxyNftOrderBo nftOrderBo = GalaxyNftOrderBo.getNew();
+            BeanUtil.copy(nftOrderVo,nftOrderBo);
+            redisUtil.set(GalaxyConstant.REDIS_KEY_GALAXY_TRADE.concat(routerType).concat(":") + nftOrderPayId, nftOrderBo);
+            return nftOrderBo;
+        }
+    }
+
+    public GalaxySeriesNftInfoBo getSeriesNftInfoBo(String routerType, String skuId) {
+        Object obj = redisUtil.get(GalaxyConstant.REDIS_KEY_GALAXY_SERIES_NFT.concat(routerType).concat(":") + skuId);
+        if(obj!=null){
+            return (GalaxySeriesNftInfoBo) obj;
+        }else {
+            GalaxySeriesNftInfoVo seriesNftInfoVo = mongoTemplate.findOne(Query.query(Criteria.where("skuId").is(skuId)), GalaxySeriesNftInfoVo.class, GalaxySeriesNftInfoVo.class.getSimpleName());
+            if (seriesNftInfoVo == null) {
+                return null;
+            }
+            GalaxySeriesNftInfoBo seriesNftInfoBo = GalaxySeriesNftInfoBo.getNew();
+            BeanUtil.copy(seriesNftInfoVo,seriesNftInfoBo);
+            redisUtil.set(GalaxyConstant.REDIS_KEY_GALAXY_SERIES_NFT.concat(routerType).concat(":") + skuId, seriesNftInfoBo);
+            return seriesNftInfoBo;
+        }
+    }
+
+    public void updateNftOrderBuyTaskId(String routerType,String nftOrderPayId,GalaxyNftOrderBo nftOrderBo) {
+        redisUtil.set(GalaxyConstant.REDIS_KEY_GALAXY_TRADE.concat(routerType).concat(":") + nftOrderPayId,nftOrderBo,keyExpireTime);
+
+        Query query = Query.query(Criteria.where("nftOrderPayId").is(nftOrderPayId));
+        Update update = Update.fromDocument(Document.parse(JsonUtils.toJson(nftOrderBo)));
+        update.set("nftBuyTaskId", nftOrderBo.getNftBuyTaskId());
+        String nowTimeStr = DateUtil.Formatter.yyyyMMddHHmmss.format(LocalDateTime.now());
+        update.set("updatedAt",nowTimeStr);
+        UpdateResult result = mongoTemplate.updateFirst(query,update, GalaxyNftOrderVo.class,GalaxyNftOrderVo.class.getSimpleName());
+        log.info("updateNftOrderInfo result:{}",result.toString());
+
+        try{
+            queueUtil.sendMySqlRedis(
+                    SqlMapping.get("galaxy_nft_order_info.updateNftBuyTaskId"),
+                    new Object[]{nftOrderBo.getNftBuyTaskId(),LocalDateTime.now(),nftOrderBo.getNftOrderPayId()}
+                    , MQConst.GalaxyQueue.SQL_NFT_TRADE_INFO.getKey()
+            );
+        }catch(Exception e){
+            log.error(e.getMessage(),e);
+            log.error("#setNftTradeInfo error ==> MESSAGE:{}",e.getMessage());
+        }
+    }
+
+}

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/Utils/KylinUtils.java → liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/utils/KylinUtils.java

-package com.liquidnet.service.consumer.kylin.Utils;
+package com.liquidnet.service.consumer.kylin.utils;
 
 import com.liquidnet.common.cache.redis.util.RedisUtil;
 import com.liquidnet.service.kylin.constant.KylinRedisConst;

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/utils/QueueUtil.java

+package com.liquidnet.service.consumer.kylin.utils;
+
+import com.liquidnet.service.base.SqlMapping;
+import com.liquidnet.service.galaxy.utils.ObjectUtil;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.connection.stream.MapRecord;
+import org.springframework.data.redis.connection.stream.StreamRecords;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.stereotype.Component;
+
+import java.util.HashMap;
+import java.util.LinkedList;
+
+/**
+ * @author AnJiabin <anjiabin@zhengzai.tv>
+ * @version V1.0
+ * @Description: TODO
+ * @class: QueueUtil
+ * @Package com.liquidnet.service.galaxy.utils
+ * @Copyright: LightNet @ Copyright (c) 2022
+ * @date 2022/3/25 15:35
+ */
+@Component
+public class QueueUtil {
+    @Autowired
+    StringRedisTemplate stringRedisTemplate;
+
+    /**
+     * 给 REDIS 队列发送消息 数据库相关
+     *
+     * @param sql  sql语句
+     * @param data 需要操作的数据
+     * @return
+     */
+    public boolean sendMySqlRedis(String sql, Object[] data , String redisKey) {
+        try {
+            LinkedList<String> sqls = ObjectUtil.cloneLinkedListStr();
+            sqls.add(sql);
+            LinkedList<Object[]> sqlsData = ObjectUtil.cloneLinkedListObj();
+            sqlsData.add(data);
+            String sqlData = SqlMapping.gets(sqls, sqlsData);
+
+            HashMap<String, String> map = ObjectUtil.cloneHashMapStringAndString();
+            map.put("message", sqlData);
+            MapRecord<String, String, String> record = StreamRecords.mapBacked(map).withStreamKey(redisKey);
+            stringRedisTemplate.opsForStream().add(record);
+            return true;
+        } catch (Exception e) {
+            e.printStackTrace();
+            return false;
+        }
+    }
+}

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/resources/bootstrap-dev.yml

@@ -6,7 +6,8 @@ liquidnet:
     username: user
     password: user123
   eureka:
-    host: 39.107.71.112:7001
+    host: 172.17.192.42:7001
+  #    host: 127.0.0.1:7001
 # end-dev-这里是配置信息基本值
 
 spring:

liquidnet-bus-service/liquidnet-service-galaxy/liquidnet-service-galaxy-impl/src/main/resources/bootstrap-dev.yml

@@ -6,8 +6,8 @@ liquidnet:
     username: user
     password: user123
   eureka:
-#    host: 172.17.192.42:7001
-    host: 127.0.0.1:7001
+    host: 172.17.192.42:7001
+#    host: 127.0.0.1:7001
 # end-dev-这里是配置信息基本值
 
 spring: