fix: 修复轮询系统缓存不一致和可观测性问题
Some checks failed
构建并部署到测试环境(无 SSH) / build-and-deploy (push) Has been cancelled
Some checks failed
构建并部署到测试环境(无 SSH) / build-and-deploy (push) Has been cancelled
- OnCardStatusChanged/Enabled/Disabled 添加 InvalidateCardCache, 解决 Refresh API 更新 DB 后 polling 缓存仍为旧值的 bug - AssetResolveResponse 的 enable_polling/network_status 去掉 omitempty,解决 false/0 时字段从响应中消失的问题 - Scheduler/Initializer/PackageHandler 增加 INFO 级别日志, 可通过日志判断轮询是否工作、处理了哪些卡
This commit is contained in:
@@ -199,6 +199,8 @@ func (p *PollingInitializer) initBatch(ctx context.Context, cards []*model.IotCa
|
||||
cardCacheTTL := 7 * 24 * time.Hour
|
||||
pipe := p.redis.Pipeline()
|
||||
cmdCount := 0
|
||||
enqueuedCards := 0
|
||||
skippedCards := 0
|
||||
|
||||
flushPipe := func() {
|
||||
if cmdCount == 0 {
|
||||
@@ -214,8 +216,10 @@ func (p *PollingInitializer) initBatch(ctx context.Context, cards []*model.IotCa
|
||||
for _, card := range cards {
|
||||
cfg := p.configMgr.MatchConfig(card)
|
||||
if cfg == nil {
|
||||
skippedCards++
|
||||
continue
|
||||
}
|
||||
enqueuedCards++
|
||||
|
||||
shardID := int(card.ID) % p.queueMgr.shardCount
|
||||
cardIDStr := fmt.Sprintf("%d", card.ID)
|
||||
@@ -273,6 +277,10 @@ func (p *PollingInitializer) initBatch(ctx context.Context, cards []*model.IotCa
|
||||
}
|
||||
|
||||
flushPipe()
|
||||
p.logger.Info("批量初始化完成",
|
||||
zap.Int("total", len(cards)),
|
||||
zap.Int("enqueued", enqueuedCards),
|
||||
zap.Int("skipped_no_config", skippedCards))
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
@@ -63,11 +63,14 @@ func (s *PollingLifecycleService) OnBatchCardsCreated(ctx context.Context, cards
|
||||
}
|
||||
}
|
||||
|
||||
// OnCardStatusChanged 卡状态变化后重新匹配配置并更新队列
|
||||
// OnCardStatusChanged 卡状态变化后清理缓存、重新匹配配置并更新队列
|
||||
func (s *PollingLifecycleService) OnCardStatusChanged(ctx context.Context, cardID uint) {
|
||||
if err := s.queueMgr.RemoveFromAllQueues(ctx, cardID); err != nil {
|
||||
s.logger.Warn("卡状态变化:从队列移除失败", zap.Uint("card_id", cardID), zap.Error(err))
|
||||
}
|
||||
// 清理轮询卡缓存,确保下次轮询从 DB 读取最新状态
|
||||
s.queueMgr.InvalidateCardCache(ctx, cardID)
|
||||
|
||||
card, err := s.iotCardStore.GetByID(ctx, cardID)
|
||||
if err != nil {
|
||||
s.logger.Error("卡状态变化:加载卡信息失败", zap.Uint("card_id", cardID), zap.Error(err))
|
||||
@@ -88,6 +91,7 @@ func (s *PollingLifecycleService) OnCardDeleted(ctx context.Context, cardID uint
|
||||
|
||||
// OnCardEnabled 卡启用轮询后初始化
|
||||
func (s *PollingLifecycleService) OnCardEnabled(ctx context.Context, cardID uint) {
|
||||
s.queueMgr.InvalidateCardCache(ctx, cardID)
|
||||
card, err := s.iotCardStore.GetByID(ctx, cardID)
|
||||
if err != nil {
|
||||
s.logger.Error("卡启用:加载卡信息失败", zap.Uint("card_id", cardID), zap.Error(err))
|
||||
@@ -104,6 +108,7 @@ func (s *PollingLifecycleService) OnCardDisabled(ctx context.Context, cardID uin
|
||||
if err := s.queueMgr.RemoveFromAllQueues(ctx, cardID); err != nil {
|
||||
s.logger.Warn("卡禁用:从队列移除失败", zap.Uint("card_id", cardID), zap.Error(err))
|
||||
}
|
||||
s.queueMgr.InvalidateCardCache(ctx, cardID)
|
||||
}
|
||||
|
||||
// shouldEnqueue M3 修复:检查卡或其绑定设备是否允许轮询
|
||||
|
||||
@@ -129,6 +129,14 @@ func (m *PollingQueueManager) OnCardDeleted(ctx context.Context, cardID uint) er
|
||||
return m.redis.Del(ctx, cacheKey).Err()
|
||||
}
|
||||
|
||||
// InvalidateCardCache 清理轮询卡信息缓存,强制下次轮询从 DB 重建
|
||||
func (m *PollingQueueManager) InvalidateCardCache(ctx context.Context, cardID uint) {
|
||||
cacheKey := constants.RedisPollingCardInfoKey(cardID)
|
||||
if err := m.redis.Del(ctx, cacheKey).Err(); err != nil {
|
||||
m.logger.Warn("清理轮询卡缓存失败", zap.Uint("card_id", cardID), zap.Error(err))
|
||||
}
|
||||
}
|
||||
|
||||
// GetQueueDepth 获取分片队列深度(用于背压检测)
|
||||
func (m *PollingQueueManager) GetQueueDepth(ctx context.Context, shardID int, taskType string) (int64, error) {
|
||||
key := constants.RedisPollingShardQueueKey(shardID, taskType)
|
||||
|
||||
@@ -186,6 +186,9 @@ func (s *Scheduler) processOneShard(ctx context.Context, shardID int) {
|
||||
for i, e := range entries {
|
||||
cardIDs[i] = formatUint(e.CardID)
|
||||
}
|
||||
s.logger.Info("分片出队",
|
||||
zap.Int("shard_id", shardID), zap.String("task_type", taskType),
|
||||
zap.Int("count", len(entries)))
|
||||
s.enqueueBatch(ctx, taskType, cardIDs)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user