From 8eb38bf162f0e063f282d6fefd505a031d7dda06 Mon Sep 17 00:00:00 2001 From: qian Date: Sat, 23 May 2026 14:46:08 +0800 Subject: [PATCH] =?UTF-8?q?improve(db):=E4=BC=98=E5=8C=96=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E5=BA=93=E9=93=BE=E6=8E=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/cmd/cmd.go | 6 ++++++ internal/cmd/cmd_migrate.go | 10 ++++++++++ internal/logic/common/config.go | 16 ++++++++++++---- internal/logic/common/content_filter.go | 16 ++++++++++++---- 4 files changed, 40 insertions(+), 8 deletions(-) diff --git a/internal/cmd/cmd.go b/internal/cmd/cmd.go index 7797a5e..4565da3 100644 --- a/internal/cmd/cmd.go +++ b/internal/cmd/cmd.go @@ -49,6 +49,12 @@ var ( return err } + // Health check: Redis + if err := checkRedis(ctx); err != nil { + g.Log().Fatalf(ctx, "Redis 连接检查失败，程序退出: %v", err) + } + g.Log().Info(ctx, "Redis 连接正常") + // Initialize JWT secret common.InitJWTSecret(ctx) diff --git a/internal/cmd/cmd_migrate.go b/internal/cmd/cmd_migrate.go index d2e705c..b7cf604 100644 --- a/internal/cmd/cmd_migrate.go +++ b/internal/cmd/cmd_migrate.go @@ -81,3 +81,13 @@ func runAutoMigrate(ctx context.Context) error { g.Log().Info(ctx, "数据库迁移完成") return nil } + +// checkRedis verifies Redis connectivity at startup with a PING command. +// Returns error if Redis is unreachable, causing the program to exit (FATA). +func checkRedis(ctx context.Context) error { + _, err := g.Redis().Do(ctx, "PING") + if err != nil { + return fmt.Errorf("Redis PING 失败: %w", err) + } + return nil +} diff --git a/internal/logic/common/config.go b/internal/logic/common/config.go index a11f254..7357dbf 100644 --- a/internal/logic/common/config.go +++ b/internal/logic/common/config.go @@ -310,20 +310,28 @@ func (s *ConfigService) Warmup(ctx context.Context) { // Called once at startup in cmd.go. func (s *ConfigService) StartSubscriber(ctx context.Context) { go func() { + var reconnectCount int for { conn, _, err := g.Redis().Subscribe(ctx, "settings:changed") if err != nil { - g.Log().Errorf(ctx, "settings subscriber connect failed: %v", err) + reconnectCount++ + g.Log().Warningf(ctx, "[PubSub:settings] 连接失败 (第%d次): %v", reconnectCount, err) time.Sleep(5 * time.Second) continue } - g.Log().Info(ctx, "settings Pub/Sub subscriber started") + if reconnectCount > 0 { + g.Log().Infof(ctx, "[PubSub:settings] 重连成功 (此前失败%d次)", reconnectCount) + reconnectCount = 0 + } else { + g.Log().Info(ctx, "[PubSub:settings] 订阅已启动") + } for { v, err := conn.Receive(ctx) if err != nil { - g.Log().Warningf(ctx, "settings subscriber recv error: %v", err) + reconnectCount++ + g.Log().Warningf(ctx, "[PubSub:settings] 接收错误 (第%d次): %v", reconnectCount, err) time.Sleep(5 * time.Second) break // reconnect } @@ -337,7 +345,7 @@ func (s *ConfigService) StartSubscriber(ctx context.Context) { s.cache.Delete(ctx, key) s.cache.Delete(ctx, "public_options") s.cache.Delete(ctx, "category:"+getCategoryForKey(key)) - g.Log().Debugf(ctx, "settings invalidated via Pub/Sub: %s", key) + g.Log().Debugf(ctx, "[PubSub:settings] 缓存已失效: %s", key) } conn.Close(ctx) diff --git a/internal/logic/common/content_filter.go b/internal/logic/common/content_filter.go index 706b342..48a7f2c 100644 --- a/internal/logic/common/content_filter.go +++ b/internal/logic/common/content_filter.go @@ -181,20 +181,28 @@ func (e *ContentFilterEngine) GetMode() string { // startSubscriber subscribes to Redis Pub/Sub "settings:changed" channel // and rebuilds the matcher when content filter settings change. func (e *ContentFilterEngine) startSubscriber(ctx context.Context) { + var reconnectCount int for { conn, _, err := g.Redis().Subscribe(ctx, "settings:changed") if err != nil { - g.Log().Errorf(ctx, "[ContentFilter] subscriber connect failed: %v", err) + reconnectCount++ + g.Log().Warningf(ctx, "[PubSub:content_filter] 连接失败 (第%d次): %v", reconnectCount, err) time.Sleep(5 * time.Second) continue } - g.Log().Info(ctx, "[ContentFilter] Pub/Sub subscriber started") + if reconnectCount > 0 { + g.Log().Infof(ctx, "[PubSub:content_filter] 重连成功 (此前失败%d次)", reconnectCount) + reconnectCount = 0 + } else { + g.Log().Info(ctx, "[PubSub:content_filter] 订阅已启动") + } for { v, err := conn.Receive(ctx) if err != nil { - g.Log().Warningf(ctx, "[ContentFilter] subscriber recv error: %v", err) + reconnectCount++ + g.Log().Warningf(ctx, "[PubSub:content_filter] 接收错误 (第%d次): %v", reconnectCount, err) time.Sleep(5 * time.Second) break // reconnect } @@ -208,7 +216,7 @@ func (e *ContentFilterEngine) startSubscriber(ctx context.Context) { switch key { case "content_filter_words", "content_filter_mode", "content_filter_replacement": e.Rebuild(ctx) - g.Log().Infof(ctx, "[ContentFilter] rebuilt via Pub/Sub for key: %s", key) + g.Log().Infof(ctx, "[PubSub:content_filter] 已重建过滤器: %s", key) } }