improve(db):优化数据库链接
This commit is contained in:
1 parent
3aca279e1b
commit
8eb38bf162
4 files changed
+40
-8
No files matched your search
@@ -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)
|
||||
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user