1214181e78
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 16m2s
- 轮询步骤异步模式下只调网关并发布 card.ingest 入库消息，入库消费者固定并发 30、按采集 ID 去重、可重试错误不限次数退避重投 - 采集记录经 card.collection 攒批写入，失败整批重投，新增 (observation_id, observed_at) 唯一索引 - 网络、实名观测增加乱序保护；轮询读查询走缓存，套餐状态查询各自计时且失败不缓存 - 入库积压告警与反压；系统配置 polling.ingest.mode 默认 sync 可回退；测试环境 nsqd 消息全部落盘 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
105 lines
3.5 KiB
Go
105 lines
3.5 KiB
Go
package task
|
||
|
||
import (
|
||
"context"
|
||
"sync"
|
||
"time"
|
||
|
||
"go.uber.org/zap"
|
||
|
||
"github.com/break/junhong_cmp_fiber/internal/model"
|
||
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
||
)
|
||
|
||
// SetIngest 注入入库消息发布器与入库模式读取（受控系统配置 polling.ingest.mode，数秒内生效）。
|
||
func (b *PollingBase) SetIngest(publisher *IngestPublisher, async func(ctx context.Context) bool) {
|
||
b.ingest = publisher
|
||
b.ingestAsync = async
|
||
}
|
||
|
||
// publishIngest 在异步入库模式下发布入库消息，返回是否已发布；返回 false 时调用方按原逻辑同步入库。
|
||
// 本轮同时是优先执行时保持同步入库：加急结果需在本轮写库后才能完成优先项。
|
||
func (b *PollingBase) publishIngest(ctx context.Context, round *pollingPriorityRound, msg IngestMessage) bool {
|
||
if round != nil || b.ingest == nil || b.ingestAsync == nil || !b.ingestAsync(ctx) {
|
||
return false
|
||
}
|
||
if err := b.ingest.PublishIngest(ctx, msg); err != nil {
|
||
b.logger.Warn("发布入库消息失败，本次改为同步入库", zap.String("kind", msg.Kind), zap.Uint("card_id", msg.CardID),
|
||
zap.String("observation_id", msg.ObservationID), zap.Error(err))
|
||
return false
|
||
}
|
||
return true
|
||
}
|
||
|
||
// scopeCache 是卡轮询范围校验结果的进程内短缓存，只供按运营商账户调度的卡任务使用，减少每个卡任务查绑定与设备。
|
||
// 设备停用轮询最迟在缓存时长后生效；生命周期事件仍按实时校验移出队列。
|
||
type scopeCache struct {
|
||
mu sync.Mutex
|
||
entries map[uint]scopeEntry
|
||
}
|
||
|
||
type scopeEntry struct {
|
||
allowed bool
|
||
loadedAt time.Time
|
||
}
|
||
|
||
// scopeAllowsCached 返回卡是否在轮询范围内（卡自身开关实时判断，设备开关按短缓存）。
|
||
func (b *PollingBase) scopeAllowsCached(ctx context.Context, card *model.IotCard) bool {
|
||
if !card.EnablePolling {
|
||
return false
|
||
}
|
||
if b.scopeChecker == nil || card.IsStandalone {
|
||
return true
|
||
}
|
||
b.scopeCache.mu.Lock()
|
||
entry, ok := b.scopeCache.entries[card.ID]
|
||
b.scopeCache.mu.Unlock()
|
||
if ok && time.Since(entry.loadedAt) < constants.PollingScopeCacheTTL {
|
||
return entry.allowed
|
||
}
|
||
allowed := b.scopeChecker.ShouldEnqueue(ctx, card)
|
||
b.scopeCache.mu.Lock()
|
||
if b.scopeCache.entries == nil || len(b.scopeCache.entries) > 200000 {
|
||
b.scopeCache.entries = map[uint]scopeEntry{}
|
||
}
|
||
b.scopeCache.entries[card.ID] = scopeEntry{allowed: allowed, loadedAt: time.Now()}
|
||
b.scopeCache.mu.Unlock()
|
||
return allowed
|
||
}
|
||
|
||
// resetDayCache 是运营商账户流量重置日的进程内缓存。
|
||
type resetDayCache struct {
|
||
mu sync.Mutex
|
||
entries map[uint]resetDayEntry
|
||
}
|
||
|
||
type resetDayEntry struct {
|
||
day int
|
||
loadedAt time.Time
|
||
}
|
||
|
||
// carrierResetDay 返回运营商账户流量重置日（1-28），成功读取的值按 1 分钟缓存；读取失败或未配置时返回 1 且不缓存。
|
||
func (b *PollingBase) carrierResetDay(ctx context.Context, carrierID uint) int {
|
||
b.resetDays.mu.Lock()
|
||
entry, ok := b.resetDays.entries[carrierID]
|
||
b.resetDays.mu.Unlock()
|
||
if ok && time.Since(entry.loadedAt) < constants.PollingCarrierResetDayCacheTTL {
|
||
return entry.day
|
||
}
|
||
if b.db == nil {
|
||
return 1
|
||
}
|
||
var days []int
|
||
if err := b.db.WithContext(ctx).Unscoped().Model(&model.Carrier{}).Where("id = ?", carrierID).Limit(1).
|
||
Pluck("data_reset_day", &days).Error; err != nil || len(days) == 0 || days[0] == 0 {
|
||
return 1
|
||
}
|
||
b.resetDays.mu.Lock()
|
||
if b.resetDays.entries == nil {
|
||
b.resetDays.entries = map[uint]resetDayEntry{}
|
||
}
|
||
b.resetDays.entries[carrierID] = resetDayEntry{day: days[0], loadedAt: time.Now()}
|
||
b.resetDays.mu.Unlock()
|
||
return days[0]
|
||
}
|