From 647d78309dce3323e063d8636c9ce7783eaa8268 Mon Sep 17 00:00:00 2001 From: huang Date: Wed, 15 Apr 2026 12:22:42 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=B0=86=E5=8D=A1=E7=8A=B6=E6=80=81?= =?UTF-8?q?=E4=BB=BB=E5=8A=A1=E7=B1=BB=E5=9E=8B=E6=8E=A5=E5=85=A5=E9=98=9F?= =?UTF-8?q?=E5=88=97=E7=AE=A1=E7=90=86=E3=80=81=E7=94=9F=E5=91=BD=E5=91=A8?= =?UTF-8?q?=E6=9C=9F=E5=92=8C=E5=88=9D=E5=A7=8B=E5=8C=96=E5=99=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit queue_manager.go: allTaskTypes 追加 TaskTypePollingCardStatus，注释更正为5个队列 lifecycle_service.go: getEnabledTaskTypes 和 calcInitialDelay 新增 card_status 条件 initializer.go: initBatch 新增 CardStatusCheckInterval 块，以 LastCardStatusCheckAt 为基准写入分片队列 Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus --- internal/polling/initializer.go | 7 +++++++ internal/polling/lifecycle_service.go | 7 +++++++ internal/polling/queue_manager.go | 3 ++- 3 files changed, 16 insertions(+), 1 deletion(-) diff --git a/internal/polling/initializer.go b/internal/polling/initializer.go index ded6b78..5507046 100644 --- a/internal/polling/initializer.go +++ b/internal/polling/initializer.go @@ -252,6 +252,13 @@ func (p *PollingInitializer) initBatch(ctx context.Context, cards []*model.IotCa }) cmdCount++ } + if cfg.CardStatusCheckInterval != nil && *cfg.CardStatusCheckInterval > 0 { + nextCheck := calculateNextCheckTime(card.LastCardStatusCheckAt, *cfg.CardStatusCheckInterval, now) + pipe.ZAdd(ctx, constants.RedisPollingShardQueueKey(shardID, constants.TaskTypePollingCardStatus), redis.Z{ + Score: float64(nextCheck.Unix()), Member: cardIDStr, + }) + cmdCount++ + } cacheKey := constants.RedisPollingCardInfoKey(card.ID) cacheData := map[string]interface{}{ diff --git a/internal/polling/lifecycle_service.go b/internal/polling/lifecycle_service.go index 5f558b4..40bcd54 100644 --- a/internal/polling/lifecycle_service.go +++ b/internal/polling/lifecycle_service.go @@ -162,6 +162,9 @@ func getEnabledTaskTypes(cfg *model.PollingConfig) []string { if cfg.ProtectCheckInterval != nil && *cfg.ProtectCheckInterval > 0 { types = append(types, constants.TaskTypePollingProtect) } + if cfg.CardStatusCheckInterval != nil && *cfg.CardStatusCheckInterval > 0 { + types = append(types, constants.TaskTypePollingCardStatus) + } return types } @@ -188,6 +191,10 @@ func calcInitialDelay(_ *model.IotCard, cfg *model.PollingConfig, taskType strin if cfg.ProtectCheckInterval != nil { interval = *cfg.ProtectCheckInterval } + case constants.TaskTypePollingCardStatus: + if cfg.CardStatusCheckInterval != nil { + interval = *cfg.CardStatusCheckInterval + } } if interval <= 0 { return now diff --git a/internal/polling/queue_manager.go b/internal/polling/queue_manager.go index d175f8e..7275893 100644 --- a/internal/polling/queue_manager.go +++ b/internal/polling/queue_manager.go @@ -18,6 +18,7 @@ var allTaskTypes = []string{ constants.TaskTypePollingCarddata, constants.TaskTypePollingPackage, constants.TaskTypePollingProtect, + constants.TaskTypePollingCardStatus, } // dequeueScript Lua 脚本：原子出队（ZRANGEBYSCORE + ZREM 服务端原子执行） @@ -97,7 +98,7 @@ func (m *PollingQueueManager) Requeue(ctx context.Context, cardID uint, taskType }).Err() } -// RemoveFromAllQueues 从所有分片的所有4个队列（含protect）移除指定卡 +// RemoveFromAllQueues 从所有分片的所有5个队列（realname/carddata/package/protect/card_status）移除指定卡 // 修复 Bug3：旧实现漏掉 protect 队列 func (m *PollingQueueManager) RemoveFromAllQueues(ctx context.Context, cardID uint) error { member := fmt.Sprintf("%d", cardID)