Merge remote-tracking branch 'origin/consumer_unite' into consumer_unite

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/config/ConsumerCommonSqlRedisStreamConfig.java

@@ -3,6 +3,7 @@ package com.liquidnet.service.consumer.base.config;
 import com.liquidnet.common.cache.redis.config.RedisStreamConfig;
 import com.liquidnet.service.base.constant.MQConst;
 import com.liquidnet.service.consumer.base.receiver.*;
+import com.liquidnet.service.consumer.base.receiver.kylin.ConsumerKylinOrderRefundExpressRdsReceiver;
 import com.liquidnet.service.consumer.base.receiver.sweet.*;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
@@ -52,6 +53,8 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
     /*------kylin------*/
     @Autowired
     ConsumerKylinSqlStationReceiver consumerKylinSqlStationReceiver;
+    @Autowired
+    ConsumerKylinOrderRefundExpressRdsReceiver consumerKylinOrderRefundExpressRdsReceiver;
     /*------slime------*/
     @Autowired
     ConsumerSlimeSqlFieldReceiver consumerSlimeSqlFieldReceiver;
@@ -408,6 +411,21 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
         }
         return subscriptionList;
     }
+    @Bean// 用户退款发快递
+    public List<Subscription> subscriptionKylinSqlOrderRefundExpress(RedisConnectionFactory factory) {
+        List<Subscription> subscriptionList = new ArrayList<>();
+        MQConst.KylinQueue stream = MQConst.KylinQueue.SQL_ORDER_REFUND_EXPRESS;
+        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()), consumerKylinOrderRefundExpressRdsReceiver
+            ));
+            listenerContainer.start();
+        }
+        return subscriptionList;
+    }
 
     /*------slime------*/
 

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

+package com.liquidnet.service.consumer.base.receiver.kylin;
+
+import com.liquidnet.service.base.constant.MQConst;
+import com.liquidnet.service.consumer.base.receiver.AbstractSqlRedisReceiver;
+import org.springframework.stereotype.Component;
+
+@Component
+public class ConsumerKylinOrderRefundExpressRdsReceiver extends AbstractSqlRedisReceiver {
+    @Override
+    protected String getRedisStreamKey() {
+        return MQConst.KylinQueue.SQL_ORDER_REFUND_EXPRESS.getKey();
+    }
+
+    @Override
+    protected String getRedisStreamGroup() {
+        return MQConst.KylinQueue.SQL_ORDER_REFUND_EXPRESS.getGroup();
+    }
+}
\ No newline at end of file