csxj2026/junhong_cmp_fiber
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects 1 Releases Wiki Activity

修复卡轮训的问题
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 10m14s
Details

Browse Source
This commit is contained in:
huang
2026-04-26 12:41:16 +08:00
parent 7efcea2b54
commit a887f91686
5 changed files with 34 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
internal/task/polling_card_cache.go
View File

@@ -46,6 +46,14 @@ func (b *PollingBase) updateCardCache(ctx context.Context, cardID uint, updates
}
}
// invalidateCardCache 删除轮询卡缓存，确保后续从数据库重建
func (b *PollingBase) invalidateCardCache(ctx context.Context, cardID uint) {
key := constants.RedisPollingCardInfoKey(cardID)
if err := b.redis.Del(ctx, key).Err(); err != nil {
b.logger.Warn("删除卡缓存失败", zap.Uint("card_id", cardID), zap.Error(err))
}
}
// parseCardFromCache 从 Redis Hash 反序列化卡信息
func parseCardFromCache(cardID uint, result map[string]string) *model.IotCard {
card := &model.IotCard{}

3
internal/task/polling_cardstatus_handler.go
View File

@@ -104,6 +104,9 @@ func (h *PollingCardStatusHandler) Handle(ctx context.Context, t *asynq.Task) er
}
if updateErr := h.iotCardStore.UpdateFields(ctx, cardID, fields); updateErr != nil {
h.base.logger.Warn("更新卡状态信息失败", zap.Uint("card_id", cardID), zap.Error(updateErr))
h.base.invalidateCardCache(ctx, cardID)
h.base.updateStats(ctx, constants.TaskTypePollingCardStatus, false, time.Since(startTime))
return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCardStatus)
}
if statusChanged {

6
internal/task/polling_protect_handler.go
View File

@@ -107,6 +107,9 @@ func (h *PollingProtectHandler) Handle(ctx context.Context, t *asynq.Task) error
"stop_reason": constants.StopReasonProtectPeriod,
}); updateErr != nil {
h.base.logger.Warn("保护期停机 DB 更新失败", zap.Uint("card_id", cardID), zap.Error(updateErr))
h.base.invalidateCardCache(ctx, cardID)
h.base.updateStats(ctx, constants.TaskTypePollingProtect, false, time.Since(startTime))
return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingProtect)
}
h.base.updateCardCache(ctx, cardID, map[string]any{"network_status": constants.NetworkStatusOffline})
actionTaken = "forced_stop"
@@ -130,6 +133,9 @@ func (h *PollingProtectHandler) Handle(ctx context.Context, t *asynq.Task) error
"stop_reason": "",
}); updateErr != nil {
h.base.logger.Warn("保护期复机 DB 更新失败", zap.Uint("card_id", cardID), zap.Error(updateErr))
h.base.invalidateCardCache(ctx, cardID)
h.base.updateStats(ctx, constants.TaskTypePollingProtect, false, time.Since(startTime))
return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingProtect)
}
h.base.updateCardCache(ctx, cardID, map[string]any{"network_status": constants.NetworkStatusOnline})
actionTaken = "forced_resume"

3
internal/task/polling_realname_handler.go
View File

@@ -129,6 +129,9 @@ func (h *PollingRealnameHandler) Handle(ctx context.Context, t *asynq.Task) erro
}
if err := h.iotCardStore.UpdateFields(ctx, cardID, fields); err != nil {
h.base.logger.Error("更新卡实名信息失败", zap.Uint("card_id", cardID), zap.Error(err))
h.base.invalidateCardCache(ctx, cardID)
h.base.updateStats(ctx, constants.TaskTypePollingRealname, false, time.Since(startTime))
return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingRealname)
}
if statusChanged {