荣耀

src/main/java/com/adx/tencent/AppConfiguration.java

@@ -40,6 +40,7 @@ import org.springframework.context.annotation.Configuration;
 import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
+import org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration;
 import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Component;
@@ -84,7 +85,17 @@ public class AppConfiguration {
         if (props.getRedisPassword() != null && !props.getRedisPassword().isBlank()) {
             cfg.setPassword(props.getRedisPassword());
         }
-        return new LettuceConnectionFactory(cfg);
+
+        LettuceClientConfiguration clientCfg = LettuceClientConfiguration.builder()
+                .commandTimeout(props.getRedisCommandTimeout())
+                .shutdownTimeout(props.getRedisShutdownTimeout())
+                .build();
+
+        LettuceConnectionFactory factory = new LettuceConnectionFactory(cfg, clientCfg);
+        factory.setValidateConnection(props.isRedisValidateConnection());
+        factory.setShareNativeConnection(false);
+        factory.setTimeout(props.getRedisConnectTimeout().toMillis());
+        return factory;
     }
 
     @Bean
@@ -148,12 +159,13 @@ public class AppConfiguration {
         if (username != null && !username.isBlank()) config.setUsername(username);
         if (password != null && !password.isBlank()) config.setPassword(password);
         config.setPoolName("adx-tidb");
-        config.setMinimumIdle(2);
-        config.setMaximumPoolSize(10);
-        config.setConnectionTimeout(Duration.ofSeconds(10).toMillis());
-        config.setValidationTimeout(Duration.ofSeconds(5).toMillis());
-        config.setIdleTimeout(Duration.ofMinutes(10).toMillis());
-        config.setMaxLifetime(Duration.ofMinutes(30).toMillis());
+        config.setMinimumIdle(props.getTidbMinimumIdle());
+        config.setMaximumPoolSize(props.getTidbMaximumPoolSize());
+        config.setConnectionTimeout(props.getTidbConnectionTimeout().toMillis());
+        config.setValidationTimeout(props.getTidbValidationTimeout().toMillis());
+        config.setIdleTimeout(props.getTidbIdleTimeout().toMillis());
+        config.setMaxLifetime(props.getTidbMaxLifetime().toMillis());
+        config.setKeepaliveTime(props.getTidbKeepaliveTime().toMillis());
         return new HikariDataSource(config);
     }
 

src/main/java/com/adx/tencent/config/AppProperties.java

@@ -81,11 +81,22 @@ public class AppProperties {
     private String redisStream = "adx:tencent:events";
     private long redisStreamMaxLen = 50000;
     private Duration bidTtl = Duration.ofDays(5);
+    private Duration redisConnectTimeout = Duration.ofSeconds(5);
+    private Duration redisCommandTimeout = Duration.ofSeconds(5);
+    private Duration redisShutdownTimeout = Duration.ofMillis(100);
+    private boolean redisValidateConnection = true;
 
     // TiDB
     private String tidbUrl;
     private String tidbUsername;
     private String tidbPassword;
+    private int tidbMinimumIdle = 2;
+    private int tidbMaximumPoolSize = 10;
+    private Duration tidbConnectionTimeout = Duration.ofSeconds(10);
+    private Duration tidbValidationTimeout = Duration.ofSeconds(3);
+    private Duration tidbIdleTimeout = Duration.ofMinutes(5);
+    private Duration tidbMaxLifetime = Duration.ofMinutes(10);
+    private Duration tidbKeepaliveTime = Duration.ofMinutes(2);
 
     // Task Lock
     private Duration taskLockTtl = Duration.ofSeconds(120);
@@ -549,6 +560,38 @@ public class AppProperties {
         this.bidTtl = v;
     }
 
+    public Duration getRedisConnectTimeout() {
+        return redisConnectTimeout;
+    }
+
+    public void setRedisConnectTimeout(Duration v) {
+        this.redisConnectTimeout = v;
+    }
+
+    public Duration getRedisCommandTimeout() {
+        return redisCommandTimeout;
+    }
+
+    public void setRedisCommandTimeout(Duration v) {
+        this.redisCommandTimeout = v;
+    }
+
+    public Duration getRedisShutdownTimeout() {
+        return redisShutdownTimeout;
+    }
+
+    public void setRedisShutdownTimeout(Duration v) {
+        this.redisShutdownTimeout = v;
+    }
+
+    public boolean isRedisValidateConnection() {
+        return redisValidateConnection;
+    }
+
+    public void setRedisValidateConnection(boolean v) {
+        this.redisValidateConnection = v;
+    }
+
     public String getTidbUrl() {
         return tidbUrl;
     }
@@ -573,6 +616,62 @@ public class AppProperties {
         this.tidbPassword = v;
     }
 
+    public int getTidbMinimumIdle() {
+        return tidbMinimumIdle;
+    }
+
+    public void setTidbMinimumIdle(int v) {
+        this.tidbMinimumIdle = v;
+    }
+
+    public int getTidbMaximumPoolSize() {
+        return tidbMaximumPoolSize;
+    }
+
+    public void setTidbMaximumPoolSize(int v) {
+        this.tidbMaximumPoolSize = v;
+    }
+
+    public Duration getTidbConnectionTimeout() {
+        return tidbConnectionTimeout;
+    }
+
+    public void setTidbConnectionTimeout(Duration v) {
+        this.tidbConnectionTimeout = v;
+    }
+
+    public Duration getTidbValidationTimeout() {
+        return tidbValidationTimeout;
+    }
+
+    public void setTidbValidationTimeout(Duration v) {
+        this.tidbValidationTimeout = v;
+    }
+
+    public Duration getTidbIdleTimeout() {
+        return tidbIdleTimeout;
+    }
+
+    public void setTidbIdleTimeout(Duration v) {
+        this.tidbIdleTimeout = v;
+    }
+
+    public Duration getTidbMaxLifetime() {
+        return tidbMaxLifetime;
+    }
+
+    public void setTidbMaxLifetime(Duration v) {
+        this.tidbMaxLifetime = v;
+    }
+
+    public Duration getTidbKeepaliveTime() {
+        return tidbKeepaliveTime;
+    }
+
+    public void setTidbKeepaliveTime(Duration v) {
+        this.tidbKeepaliveTime = v;
+    }
+
     public Duration getTaskLockTtl() {
         return taskLockTtl;
     }

src/main/java/com/adx/tencent/honor/HonorHotStore.java

@@ -13,6 +13,8 @@ import org.springframework.data.redis.connection.stream.RecordId;
 import org.springframework.data.redis.connection.stream.StreamOffset;
 import org.springframework.data.redis.connection.stream.StreamReadOptions;
 import org.springframework.data.redis.core.StringRedisTemplate;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
@@ -25,15 +27,19 @@ import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 
 public class HonorHotStore {
 
+    private static final Logger log = LoggerFactory.getLogger(HonorHotStore.class);
+
     private final StringRedisTemplate redis;
     private final ObjectMapper objectMapper;
     private final HonorColdStore coldStore;
     private final String prefix;
     private final String stream;
     private final Duration bidTtl;
+    private final Set<String> initializedGroups = ConcurrentHashMap.newKeySet();
 
     public HonorHotStore(StringRedisTemplate redis, ObjectMapper objectMapper,
                          HonorColdStore coldStore, String prefix, String stream, Duration bidTtl) {
@@ -151,17 +157,28 @@ public class HonorHotStore {
     }
 
     public void ensureGroup(String group) {
+        if (group == null || group.isBlank()) return;
+        if (initializedGroups.contains(group)) return;
         try {
             redis.opsForStream().createGroup(stream, ReadOffset.from("0"), group);
+            initializedGroups.add(group);
         } catch (Exception e) {
-            if (isBusyGroupError(e)) return;
+            if (isBusyGroupError(e)) {
+                initializedGroups.add(group);
+                return;
+            }
             try {
                 var id = redis.opsForStream().add(org.springframework.data.redis.connection.stream.StreamRecords
                         .newRecord().ofMap(Collections.singletonMap("_init", "1")).withStreamKey(stream));
                 if (id != null) redis.opsForStream().delete(stream, id);
                 redis.opsForStream().createGroup(stream, ReadOffset.from("0"), group);
+                initializedGroups.add(group);
             } catch (Exception retryEx) {
-                if (!isBusyGroupError(retryEx)) throw new RuntimeException("ensure honor group failed", retryEx);
+                if (isBusyGroupError(retryEx)) {
+                    initializedGroups.add(group);
+                    return;
+                }
+                throw new RuntimeException("ensure honor group failed", retryEx);
             }
         }
     }
@@ -176,6 +193,8 @@ public class HonorHotStore {
             return toQueuedEvents(records);
         } catch (Exception e) {
             if (e.getMessage() != null && e.getMessage().contains("NOGROUP")) return Collections.emptyList();
+            log.error("[HonorHotStore] read failed | stream={} | group={} | consumer={} | count={} | error={}",
+                    stream, group, consumer, count, rootMessage(e), e);
             throw new RuntimeException("read honor stream failed", e);
         }
     }
@@ -243,4 +262,16 @@ public class HonorHotStore {
         }
         return false;
     }
+
+    private static String rootMessage(Throwable e) {
+        Throwable cur = e;
+        String msg = null;
+        while (cur != null) {
+            if (cur.getMessage() != null && !cur.getMessage().isBlank()) {
+                msg = cur.getMessage();
+            }
+            cur = cur.getCause();
+        }
+        return msg == null ? "" : msg;
+    }
 }

src/main/java/com/adx/tencent/storage/RedisHotStore.java

@@ -11,6 +11,7 @@ import java.time.Duration;
 import java.time.Instant;
 import java.nio.charset.StandardCharsets;
 import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
 
 /**
  * 对应 Go internal/storage/redis.go 的 RedisHotStore。
@@ -32,6 +33,7 @@ public class RedisHotStore {
     private final String prefix;
     private final String stream;
     private final Duration bidTtl;
+    private final Set<String> initializedGroups = ConcurrentHashMap.newKeySet();
 
     public RedisHotStore(StringRedisTemplate redis, ObjectMapper objectMapper,
                          TiDBColdStore coldStore, String prefix, String stream, Duration bidTtl) {
@@ -206,10 +208,14 @@ public class RedisHotStore {
      * 对应 Go EnsureGroup：XGROUP CREATE MKSTREAM（忽略 BUSYGROUP 错误）。
      */
     public void ensureGroup(String group) {
+        if (group == null || group.isBlank()) return;
+        if (initializedGroups.contains(group)) return;
         try {
             redis.opsForStream().createGroup(stream, ReadOffset.from("0"), group);
+            initializedGroups.add(group);
         } catch (Exception e) {
             if (isBusyGroupError(e)) {
+                initializedGroups.add(group);
                 return; // 组已存在，忽略
             }
             // Stream 可能不存在，先创建 Stream 再重试
@@ -222,10 +228,13 @@ public class RedisHotStore {
                     redis.opsForStream().delete(stream, id);
                 }
                 redis.opsForStream().createGroup(stream, ReadOffset.from("0"), group);
+                initializedGroups.add(group);
             } catch (Exception retryEx) {
-                if (!isBusyGroupError(retryEx)) {
-                    throw new RuntimeException("ensureGroup failed", retryEx);
+                if (isBusyGroupError(retryEx)) {
+                    initializedGroups.add(group);
+                    return;
                 }
+                throw new RuntimeException("ensureGroup failed", retryEx);
             }
         }
     }

src/main/resources/application-dev.yml

@@ -45,11 +45,22 @@ adx:
   redis-username: "hcst"
   redis-password: "hcst@2026qwe.."
   redis-db: 5
+  redis-connect-timeout: 5s
+  redis-command-timeout: 5s
+  redis-shutdown-timeout: 100ms
+  redis-validate-connection: true
 
   # --- TiDB/MySQL ---
   tidb-url: "jdbc:mysql://139.186.172.149:3390/adx_tencent?useSSL=false&useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&allowPublicKeyRetrieval=true"
   tidb-username: hcst
   tidb-password: "hcst@2025"
+  tidb-minimum-idle: 2
+  tidb-maximum-pool-size: 10
+  tidb-connection-timeout: 10s
+  tidb-validation-timeout: 3s
+  tidb-idle-timeout: 5m
+  tidb-max-lifetime: 10m
+  tidb-keepalive-time: 2m
 
   # --- Task ---
   skip-leader-election: true

src/main/resources/application-prod.yml

@@ -47,11 +47,22 @@ adx:
   redis-username: "hcst"
   redis-password: "hcst@2026qwe.."
   redis-db: 5
+  redis-connect-timeout: 5s
+  redis-command-timeout: 5s
+  redis-shutdown-timeout: 100ms
+  redis-validate-connection: true
 
   # --- TiDB/MySQL ---
   tidb-url: "jdbc:mysql://172.30.0.36:3390/adx_tencent?useSSL=false&useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&allowPublicKeyRetrieval=true"
   tidb-username: "hcst"
   tidb-password: "hcst@2025"
+  tidb-minimum-idle: 2
+  tidb-maximum-pool-size: 10
+  tidb-connection-timeout: 10s
+  tidb-validation-timeout: 3s
+  tidb-idle-timeout: 5m
+  tidb-max-lifetime: 10m
+  tidb-keepalive-time: 2m
 
   # --- Task ---
   skip-leader-election: false

src/main/resources/application-test.yml

@@ -46,11 +46,22 @@ adx:
   redis-username: hcst
   redis-password: "hcst@2026qwe.."
   redis-db: 0
+  redis-connect-timeout: 5s
+  redis-command-timeout: 5s
+  redis-shutdown-timeout: 100ms
+  redis-validate-connection: true
 
   # --- TiDB/MySQL ---
   tidb-url: "jdbc:mysql://139.186.172.149:3390/adx_tencent?useSSL=false&useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&allowPublicKeyRetrieval=true&connectTimeout=5000&socketTimeout=30000"
   tidb-username: hcst
   tidb-password: "hcst@2025"
+  tidb-minimum-idle: 2
+  tidb-maximum-pool-size: 10
+  tidb-connection-timeout: 10s
+  tidb-validation-timeout: 3s
+  tidb-idle-timeout: 5m
+  tidb-max-lifetime: 10m
+  tidb-keepalive-time: 2m
 
   # --- Task ---
   skip-leader-election: false