2fe7c1f8ac
- 四个轮询处理器拆为步骤函数 Step（返回结果分类），旧 Asynq Handle 只做并发控制与原有重入队映射，行为不变；carrier 模式下旧 handler 直接结束
- 网关限流增加等待模式（ctx 标记），NSQ 消费者等名额到单步超时，超时按限流处理不计失败；记录限流等待次数
- 卡任务执行器：核对 token 并抢执行权，按配置步骤顺序串行执行到期检查（单步超时、总超时、panic 恢复、流量已触发评估时跳过套餐），执行中续期并 Touch，续期失败立即停止；收尾写回到期、连续失败、退避、延迟确认与告警
- 分发器：每秒按账户抢分发锁领取（执行上限按消费能力收紧、优先占比、按限流组剩余 QPS 估算），发布到 polling.card.{账户}，发布失败释放；消费旧优先提示与手动触发；心跳与处理中标记到期计数
- NSQ 生产者与按账户动态订阅的消费者管理器，优雅退出宽限期
- 失败告警异步写 tb_polling_failure_alert（同卡同类型一条未处理，成功转已恢复），账户窗口内告警数超阈值产生账户级告警
- 模式监听：切 carrier 时旧分片调度停止出队，加锁初始化新结构后开始领取并订阅；回 legacy 时停止消费、重建旧分片队列；生命周期（API 与 Worker）在 carrier 模式写账户队列
243 lines
8.4 KiB
Go
243 lines
8.4 KiB
Go
package polling
|
||
|
||
import (
|
||
"context"
|
||
"math/rand"
|
||
"time"
|
||
|
||
"go.uber.org/zap"
|
||
|
||
"github.com/break/junhong_cmp_fiber/internal/model"
|
||
"github.com/break/junhong_cmp_fiber/internal/store/postgres"
|
||
)
|
||
|
||
// PollingLifecycleService 卡生命周期轮询管理服务
|
||
// 替代 callbacks.go 和 api_callback.go 中的生命周期方法
|
||
// 实现 iot_card.PollingCallback 接口，两个进程（API 和 Worker）共享同一实现
|
||
// 依赖：PollingQueueManager（队列操作）+ PollingConfigManager（配置匹配）
|
||
type PollingLifecycleService struct {
|
||
queueMgr *PollingQueueManager
|
||
configMgr *PollingConfigManager
|
||
iotCardStore *postgres.IotCardStore
|
||
deviceBindingStore *postgres.DeviceSimBindingStore
|
||
deviceStore *postgres.DeviceStore
|
||
logger *zap.Logger
|
||
// carrier 按运营商账户调度接缝；当前模式为 carrier 时生命周期事件写入账户队列，不写旧分片队列。
|
||
carrier *CarrierScheduling
|
||
}
|
||
|
||
// SetCarrierScheduling 注入按运营商账户调度接缝。
|
||
func (s *PollingLifecycleService) SetCarrierScheduling(carrier *CarrierScheduling) {
|
||
s.carrier = carrier
|
||
}
|
||
|
||
// carrierActive 判断当前是否按运营商账户调度。
|
||
func (s *PollingLifecycleService) carrierActive(ctx context.Context) bool {
|
||
return s.carrier.Active(ctx)
|
||
}
|
||
|
||
// NewPollingLifecycleService 创建卡生命周期轮询管理服务
|
||
func NewPollingLifecycleService(
|
||
queueMgr *PollingQueueManager,
|
||
configMgr *PollingConfigManager,
|
||
iotCardStore *postgres.IotCardStore,
|
||
deviceBindingStore *postgres.DeviceSimBindingStore,
|
||
deviceStore *postgres.DeviceStore,
|
||
logger *zap.Logger,
|
||
) *PollingLifecycleService {
|
||
return &PollingLifecycleService{
|
||
queueMgr: queueMgr,
|
||
configMgr: configMgr,
|
||
iotCardStore: iotCardStore,
|
||
deviceBindingStore: deviceBindingStore,
|
||
deviceStore: deviceStore,
|
||
logger: logger,
|
||
}
|
||
}
|
||
|
||
// OnCardCreated 新卡创建后初始化轮询
|
||
func (s *PollingLifecycleService) OnCardCreated(ctx context.Context, card *model.IotCard) {
|
||
if card == nil {
|
||
return
|
||
}
|
||
if !s.ShouldEnqueue(ctx, card) {
|
||
s.logger.Debug("卡禁用轮询，跳过入队", zap.Uint("card_id", card.ID))
|
||
return
|
||
}
|
||
if s.carrierActive(ctx) {
|
||
s.carrier.UpsertCard(ctx, card, false)
|
||
return
|
||
}
|
||
s.enqueueCard(ctx, card)
|
||
}
|
||
|
||
// OnBatchCardsCreated 批量卡创建后批量初始化轮询
|
||
func (s *PollingLifecycleService) OnBatchCardsCreated(ctx context.Context, cards []*model.IotCard) {
|
||
for _, card := range cards {
|
||
s.OnCardCreated(ctx, card)
|
||
}
|
||
}
|
||
|
||
// OnCardStatusChanged 卡状态变化后清理缓存、重新匹配配置并更新队列
|
||
func (s *PollingLifecycleService) OnCardStatusChanged(ctx context.Context, cardID uint) {
|
||
if s.carrierActive(ctx) {
|
||
s.queueMgr.InvalidateCardCache(ctx, cardID)
|
||
s.carrierRefresh(ctx, cardID)
|
||
return
|
||
}
|
||
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))
|
||
return
|
||
}
|
||
if !s.ShouldEnqueue(ctx, card) {
|
||
return
|
||
}
|
||
s.enqueueCard(ctx, card)
|
||
}
|
||
|
||
// OnBatchCardsStatusChanged 批量卡状态变化：单次 Redis pipeline 清理 + 单次 DB 批量查询
|
||
// 替代 N 次 OnCardStatusChanged 调用，避免批量分配/回收时打满 DB 连接池
|
||
func (s *PollingLifecycleService) OnBatchCardsStatusChanged(ctx context.Context, cardIDs []uint) {
|
||
if len(cardIDs) == 0 {
|
||
return
|
||
}
|
||
if s.carrierActive(ctx) {
|
||
for _, cardID := range cardIDs {
|
||
s.queueMgr.InvalidateCardCache(ctx, cardID)
|
||
s.carrierRefresh(ctx, cardID)
|
||
}
|
||
return
|
||
}
|
||
// 单次 pipeline：批量从当前分片队列移除并清理缓存
|
||
if err := s.queueMgr.RemoveBatchFromCurrentShardQueues(ctx, cardIDs); err != nil {
|
||
s.logger.Warn("批量卡状态变化：从队列移除失败", zap.Int("count", len(cardIDs)), zap.Error(err))
|
||
}
|
||
// 单次 DB 查询：批量加载所有卡信息
|
||
cards, err := s.iotCardStore.GetByIDs(ctx, cardIDs)
|
||
if err != nil {
|
||
s.logger.Error("批量卡状态变化：加载卡信息失败", zap.Int("count", len(cardIDs)), zap.Error(err))
|
||
return
|
||
}
|
||
for _, card := range cards {
|
||
if !s.ShouldEnqueue(ctx, card) {
|
||
continue
|
||
}
|
||
s.enqueueCard(ctx, card)
|
||
}
|
||
}
|
||
|
||
// OnCardDeleted 卡删除后移除所有队列并清理缓存
|
||
func (s *PollingLifecycleService) OnCardDeleted(ctx context.Context, cardID uint) {
|
||
if s.carrierActive(ctx) {
|
||
s.carrier.RemoveCard(ctx, cardID, 0)
|
||
s.queueMgr.InvalidateCardCache(ctx, cardID)
|
||
return
|
||
}
|
||
if err := s.queueMgr.OnCardDeleted(ctx, cardID); err != nil {
|
||
s.logger.Warn("卡删除：清理队列和缓存失败", zap.Uint("card_id", cardID), zap.Error(err))
|
||
}
|
||
}
|
||
|
||
// 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))
|
||
return
|
||
}
|
||
if !s.ShouldEnqueue(ctx, card) {
|
||
return
|
||
}
|
||
if s.carrierActive(ctx) {
|
||
// 恢复轮询时各检查类型到期设为当前时间加抖动。
|
||
s.carrier.UpsertCard(ctx, card, false)
|
||
return
|
||
}
|
||
s.enqueueCard(ctx, card)
|
||
}
|
||
|
||
// OnCardDisabled 卡禁用轮询后移除所有队列
|
||
func (s *PollingLifecycleService) OnCardDisabled(ctx context.Context, cardID uint) {
|
||
if s.carrierActive(ctx) {
|
||
s.carrier.RemoveCard(ctx, cardID, 0)
|
||
s.queueMgr.InvalidateCardCache(ctx, cardID)
|
||
return
|
||
}
|
||
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 修复：检查卡或其绑定设备是否允许轮询（同时被轮询优先项的执行前范围校验复用）
|
||
// 两层检查：先查卡级 enable_polling，再查设备级 enable_polling（IsStandalone=false 时）
|
||
// 直接查 device.enable_polling，不依赖级联同步正确性，防止同步失败时生命周期事件绕过设备级禁用
|
||
func (s *PollingLifecycleService) ShouldEnqueue(ctx context.Context, card *model.IotCard) bool {
|
||
if !card.EnablePolling {
|
||
return false
|
||
}
|
||
if !card.IsStandalone && s.deviceBindingStore != nil && s.deviceStore != nil {
|
||
binding, err := s.deviceBindingStore.GetActiveBindingByCardID(ctx, card.ID)
|
||
if err == nil && binding != nil {
|
||
device, devErr := s.deviceStore.GetByID(ctx, binding.DeviceID)
|
||
if devErr == nil && !device.EnablePolling {
|
||
s.logger.Debug("设备已禁用轮询，跳过卡入队",
|
||
zap.Uint("card_id", card.ID), zap.Uint("device_id", binding.DeviceID))
|
||
return false
|
||
}
|
||
}
|
||
}
|
||
return true
|
||
}
|
||
|
||
// enqueueCard 按匹配配置将卡入队分片队列
|
||
func (s *PollingLifecycleService) enqueueCard(ctx context.Context, card *model.IotCard) {
|
||
intervals := s.configMgr.MergedTaskIntervals(card)
|
||
if len(intervals) == 0 {
|
||
return
|
||
}
|
||
for taskType, info := range intervals {
|
||
if err := s.queueMgr.Requeue(ctx, card.ID, taskType, calcInitialDelay(info.Interval)); err != nil {
|
||
s.logger.Warn("卡入队失败",
|
||
zap.Uint("card_id", card.ID), zap.String("task_type", taskType), zap.Error(err))
|
||
}
|
||
}
|
||
}
|
||
|
||
// calcInitialDelay 计算新入队卡的初始检查时间（加随机抖动避免惊群）
|
||
// 抖动范围 [0, max(interval/10, 2)) 秒；下界取 2 保证 rand.Intn 有实际随机性
|
||
// （rand.Intn(1) 永远返回 0，对 interval≤10 的配置会导致零抖动）
|
||
func calcInitialDelay(interval int) time.Time {
|
||
now := time.Now()
|
||
if interval <= 0 {
|
||
return now
|
||
}
|
||
jitterMax := interval / 10
|
||
if jitterMax < 2 {
|
||
jitterMax = 2
|
||
}
|
||
return now.Add(time.Duration(rand.Intn(jitterMax)) * time.Second)
|
||
}
|
||
|
||
// carrierRefresh 按运营商账户调度时重新匹配配置并写入账户队列：不再轮询的卡移出，其余按当前时间加抖动重排。
|
||
func (s *PollingLifecycleService) carrierRefresh(ctx context.Context, cardID uint) {
|
||
card, err := s.iotCardStore.GetByID(ctx, cardID)
|
||
if err != nil {
|
||
s.logger.Error("卡状态变化：加载卡信息失败", zap.Uint("card_id", cardID), zap.Error(err))
|
||
return
|
||
}
|
||
if !s.ShouldEnqueue(ctx, card) {
|
||
s.carrier.RemoveCard(ctx, card.ID, card.CarrierID)
|
||
return
|
||
}
|
||
s.carrier.UpsertCard(ctx, card, false)
|
||
}
|