~opt;

liquidnet-bus-common/liquidnet-common-cache/liquidnet-common-cache-redis/src/main/java/com.liquidnet.common.cache/redis/config/RedisStreamConfig.java

 package com.liquidnet.common.cache.redis.config;
 
+import lombok.extern.slf4j.Slf4j;
 import lombok.var;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Configuration;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.connection.stream.MapRecord;
+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.data.redis.core.StringRedisTemplate;
 import org.springframework.data.redis.stream.StreamMessageListenerContainer;
+import org.springframework.stereotype.Component;
 
 import java.net.InetAddress;
 import java.net.UnknownHostException;
 import java.time.Duration;
+import java.util.HashMap;
 
+@Slf4j
+@Component
 public class RedisStreamConfig {
+    @Autowired
+    StringRedisTemplate stringRedisTemplate;
+
     private String hostname;
     {
         try {
@@ -19,6 +33,24 @@ public class RedisStreamConfig {
         }
     }
 
+    public void initStream(String key, String group) {
+        try {
+            if (stringRedisTemplate.hasKey(key)) {
+                log.info("redis stream exist[{},{}]", key, group);
+            } else {
+                log.info("redis stream init[{},{}]", key, group);
+                HashMap<String, String> map = new HashMap<>();
+                map.put("message", " ");
+                StreamOperations<String, Object, Object> stringObjectObjectStreamOperations = stringRedisTemplate.opsForStream();
+                RecordId recordId = stringObjectObjectStreamOperations.add(StreamRecords.mapBacked(map).withStreamKey(key));
+                stringObjectObjectStreamOperations.createGroup(key, group);
+                stringObjectObjectStreamOperations.delete(key, recordId.getValue());
+            }
+        } catch (Exception e) {
+            log.error("Ex:redis stream init [{},{}]", key, group, e);
+        }
+    }
+
     public StreamMessageListenerContainer<String, MapRecord<String, String, String>> buildStreamMessageListenerContainer(RedisConnectionFactory factory) {
         var options = StreamMessageListenerContainer
                 .StreamMessageListenerContainerOptions

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-adam/src/main/java/com/liquidnet/service/consumer/adam/config/ConsumerAdamSqlUcenterRedisStreamConfig.java

 package com.liquidnet.service.consumer.adam.config;
 
 import com.liquidnet.common.cache.redis.config.RedisStreamConfig;
-import com.liquidnet.commons.lang.util.CollectionUtil;
 import com.liquidnet.service.consumer.adam.receiver.ConsumerAdamSqlUCenterRdsReceiver;
 import com.liquidnet.service.consumer.adam.receiver.ConsumerAdamSqlUMemberRdsReceiver;
 import com.liquidnet.service.consumer.adam.receiver.ConsumerAdamSqlURegisterRdsReceiver;
@@ -10,14 +9,14 @@ 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.*;
-import org.springframework.data.redis.core.StreamOperations;
-import org.springframework.data.redis.core.StringRedisTemplate;
+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.stream.StreamMessageListenerContainer;
 import org.springframework.data.redis.stream.Subscription;
 
 import java.util.ArrayList;
-import java.util.HashMap;
 import java.util.List;
 
 import static com.liquidnet.service.base.constant.MQConst.AdamQueue;
@@ -25,8 +24,6 @@ import static com.liquidnet.service.base.constant.MQConst.AdamQueue;
 @Slf4j
 @Configuration
 public class ConsumerAdamSqlUcenterRedisStreamConfig extends RedisStreamConfig {
-    @Autowired
-    StringRedisTemplate stringRedisTemplate;
     @Autowired
     ConsumerAdamSqlURegisterRdsReceiver consumerAdamURegisterRdsReceiver;
     @Autowired
@@ -34,23 +31,6 @@ public class ConsumerAdamSqlUcenterRedisStreamConfig extends RedisStreamConfig {
     @Autowired
     ConsumerAdamSqlUMemberRdsReceiver consumerAdamUMemberRdsReceiver;
 
-    private void initStream(String key, String group) {
-        try {
-            if (!stringRedisTemplate.hasKey(key)) {
-                HashMap<String, String> map = CollectionUtil.mapStringString();
-                map.put("message", " ");
-                StreamOperations<String, Object, Object> stringObjectObjectStreamOperations = stringRedisTemplate.opsForStream();
-                RecordId recordId = stringObjectObjectStreamOperations.add(StreamRecords.mapBacked(map).withStreamKey(key));
-                stringObjectObjectStreamOperations.createGroup(key, group);
-                stringObjectObjectStreamOperations.delete(key, recordId.getValue());
-            } else {
-                log.info("redis stream exist [{},{}]", key, group);
-            }
-        } catch (Exception e) {
-            log.error("Ex:redis stream init [{},{}]", key, group, e);
-        }
-    }
-
     @Bean// 用户注册
     public List<Subscription> subscriptionSqlURegister(RedisConnectionFactory factory) {
         List<Subscription> subscriptionList = new ArrayList<>();