提交nft订单失败日志消费

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/config/ConsumerGalaxySqlNftOrderFailLogConfig.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.ConsumerGalaxySqlNftOrderFailLogReceiver;
+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: TODO
+ * @class: ConsumerGalaxySqlNftOrderFailLogConfig
+ * @Package com.liquidnet.service.consumer.kylin.config
+ * @Copyright: LightNet @ Copyright (c) 2022
+ * @date 2022/4/7 10:57
+ */
+@Configuration
+public class ConsumerGalaxySqlNftOrderFailLogConfig extends RedisStreamConfig {
+    @Autowired
+    private ConsumerGalaxySqlNftOrderFailLogReceiver sqlNftOrderFailLogReceiver;
+
+    @Autowired
+    StringRedisTemplate stringRedisTemplate;
+
+    @Bean
+    public List<Subscription> subscriptionSqlNftOrderInfo(RedisConnectionFactory factory) {
+        List<Subscription> subscriptionList = new ArrayList<>();
+        MQConst.GalaxyQueue stream = MQConst.GalaxyQueue.SQL_NFT_ORDER_FAIL_LOG;
+
+        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()), sqlNftOrderFailLogReceiver
+            ));
+            listenerContainer.start();
+        }
+        return subscriptionList;
+    }
+}

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/receiver/ConsumerGalaxySqlNftOrderFailLogReceiver.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;
+
+/**
+ * @author AnJiabin <anjiabin@zhengzai.tv>
+ * @version V1.0
+ * @Description: TODO
+ * @class: ConsumerGalaxySqlNftOrderFailLogReceiver
+ * @Package com.liquidnet.service.consumer.kylin.receiver
+ * @Copyright: LightNet @ Copyright (c) 2022
+ * @date 2022/4/7 10:56
+ */
+@Slf4j
+@Component
+public class ConsumerGalaxySqlNftOrderFailLogReceiver extends AbstractSqlRedisReceiver {
+
+    @Override
+    protected String getRedisStreamKey() {
+        return MQConst.GalaxyQueue.SQL_NFT_ORDER_FAIL_LOG.getKey();
+    }
+
+    @Override
+    protected String getRedisStreamGroup() {
+        return MQConst.GalaxyQueue.SQL_NFT_ORDER_FAIL_LOG.getGroup();
+    }
+}