合并发行与购买

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

+package com.liquidnet.service.consumer.galaxy.config;
+
+import com.liquidnet.common.cache.redis.config.RedisStreamConfig;
+import com.liquidnet.service.base.constant.MQConst;
+import com.liquidnet.service.consumer.galaxy.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> subscriptionSqlNftOrderInfo(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-galaxy/src/main/java/com/liquidnet/service/consumer/galaxy/receiver/RedisJsonNftPublishAndBuyReceiver.java → liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-galaxy/src/main/java/com/liquidnet/service/consumer/galaxy/receiver/JsonRedisJsonNftPublishAndBuyReceiver.java

@@ -6,7 +6,7 @@ import org.springframework.stereotype.Component;
 
 @Slf4j
 @Component
-public class RedisJsonNftPublishAndBuyReceiver extends AbstractJsonRedisReceiver {
+public class JsonRedisJsonNftPublishAndBuyReceiver extends AbstractJsonRedisReceiver {
 
     @Override
     protected String getRedisStreamKey() {

liquidnet-bus-service/liquidnet-service-galaxy/liquidnet-service-galaxy-impl/src/main/java/com/liquidnet/service/galaxy/router/zxin/service/GalaxyRouterStrategyZxlTradeImpl.java

 package com.liquidnet.service.galaxy.router.zxin.service;
 
+import com.liquidnet.commons.lang.util.JsonUtils;
 import com.liquidnet.service.base.ResponseDto;
+import com.liquidnet.service.base.constant.MQConst;
 import com.liquidnet.service.galaxy.constant.GalaxyEnum;
 import com.liquidnet.service.galaxy.dto.param.*;
 import com.liquidnet.service.galaxy.router.strategy.IGalaxyRouterStrategyTrade;
 import com.liquidnet.service.galaxy.router.strategy.annotation.StrategyGalaxyRouterTradeHandler;
 import com.liquidnet.service.galaxy.router.zxin.biz.ZxinTradeBiz;
+import com.liquidnet.service.galaxy.utils.QueueUtil;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
@@ -26,9 +29,14 @@ public class GalaxyRouterStrategyZxlTradeImpl implements IGalaxyRouterStrategyTr
     @Autowired
     private ZxinTradeBiz zxinTradeBiz;
 
+    @Autowired
+    private QueueUtil queueUtil;
+
     @Override
     public ResponseDto<GalaxyNftPublishAndBuyRespDto> nftPublishAndBuy(GalaxyNftPublishAndBuyReqDto reqDto) {
-        return zxinTradeBiz.nftPublishAndBuy(reqDto);
+        queueUtil.sendMsgByRedis(MQConst.GalaxyQueue.JSON_NFT_PUBLISH_AND_BUY.getKey(), JsonUtils.toJson(reqDto));
+        return ResponseDto.success();
+//        return zxinTradeBiz.nftPublishAndBuy(reqDto);
     }
 
     @Override

liquidnet-bus-service/liquidnet-service-galaxy/liquidnet-service-galaxy-impl/src/main/java/com/liquidnet/service/galaxy/utils/QueueUtil.java

 package com.liquidnet.service.galaxy.utils;
 
+import com.liquidnet.commons.lang.util.CollectionUtil;
 import com.liquidnet.service.base.SqlMapping;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.redis.connection.stream.MapRecord;
@@ -49,4 +50,16 @@ public class QueueUtil {
             return false;
         }
     }
+
+    /**
+     * 发送消息 - REDIS
+     *
+     * @param streamKey Redis消费Key
+     * @param jsonMsg   Json字符串
+     */
+    public void sendMsgByRedis(String streamKey, String jsonMsg) {
+        HashMap<String, String> map = CollectionUtil.mapStringString();
+        map.put("message", jsonMsg);
+        stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(streamKey));
+    }
 }