huang
931e140e8e
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 6m35s
实现功能： - 实名状态检查轮询（可配置间隔） - 卡流量检查轮询（支持跨月流量追踪） - 套餐检查与超额自动停机 - 分布式并发控制（Redis 信号量） - 手动触发轮询（单卡/批量/条件筛选） - 数据清理配置与执行 - 告警规则与历史记录 - 实时监控统计（队列/性能/并发） 性能优化： - Redis 缓存卡信息，减少 DB 查询 - Pipeline 批量写入 Redis - 异步流量记录写入 - 渐进式初始化（10万卡/批） 压测工具（scripts/benchmark/）： - Mock Gateway 模拟上游服务 - 测试卡生成器 - 配置初始化脚本 - 实时监控脚本 Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
108 lines
3.2 KiB
Go
108 lines
3.2 KiB
Go
package polling
|
||
|
||
import (
|
||
"context"
|
||
"strconv"
|
||
|
||
"github.com/redis/go-redis/v9"
|
||
"go.uber.org/zap"
|
||
|
||
"github.com/break/junhong_cmp_fiber/internal/model"
|
||
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
||
)
|
||
|
||
// APICallback API 进程使用的轻量级轮询回调
|
||
// 直接操作 Redis 队列，不依赖调度器
|
||
type APICallback struct {
|
||
redis *redis.Client
|
||
logger *zap.Logger
|
||
}
|
||
|
||
// NewAPICallback 创建 API 回调实例
|
||
func NewAPICallback(redis *redis.Client, logger *zap.Logger) *APICallback {
|
||
return &APICallback{
|
||
redis: redis,
|
||
logger: logger,
|
||
}
|
||
}
|
||
|
||
// OnCardCreated 卡创建时的回调
|
||
// 注意：大多数卡创建是通过 Worker 的批量导入完成的，这个方法主要用于单卡创建场景
|
||
func (c *APICallback) OnCardCreated(ctx context.Context, card *model.IotCard) {
|
||
if card == nil {
|
||
return
|
||
}
|
||
c.logger.Debug("API 回调：卡创建", zap.Uint("card_id", card.ID))
|
||
// 卡创建后，scheduler 的渐进式初始化会将其加入队列
|
||
// 这里不做处理，让 scheduler 处理
|
||
}
|
||
|
||
// OnCardStatusChanged 卡状态变化时的回调
|
||
func (c *APICallback) OnCardStatusChanged(ctx context.Context, cardID uint) {
|
||
c.logger.Debug("API 回调：卡状态变化", zap.Uint("card_id", cardID))
|
||
// 状态变化后，scheduler 下次扫描时会更新配置匹配
|
||
// 这里不做处理，让 scheduler 处理
|
||
}
|
||
|
||
// OnCardDeleted 卡删除时的回调
|
||
// 从所有队列中移除卡
|
||
func (c *APICallback) OnCardDeleted(ctx context.Context, cardID uint) {
|
||
c.logger.Debug("API 回调：卡删除", zap.Uint("card_id", cardID))
|
||
|
||
member := strconv.FormatUint(uint64(cardID), 10)
|
||
|
||
// 从所有轮询队列中移除
|
||
queues := []string{
|
||
constants.RedisPollingQueueRealnameKey(),
|
||
constants.RedisPollingQueueCarddataKey(),
|
||
constants.RedisPollingQueuePackageKey(),
|
||
}
|
||
|
||
for _, queueKey := range queues {
|
||
if err := c.redis.ZRem(ctx, queueKey, member).Err(); err != nil {
|
||
c.logger.Warn("从队列移除卡失败",
|
||
zap.String("queue", queueKey),
|
||
zap.Uint("card_id", cardID),
|
||
zap.Error(err))
|
||
}
|
||
}
|
||
|
||
// 删除卡信息缓存
|
||
cacheKey := constants.RedisPollingCardInfoKey(cardID)
|
||
if err := c.redis.Del(ctx, cacheKey).Err(); err != nil {
|
||
c.logger.Warn("删除卡缓存失败",
|
||
zap.Uint("card_id", cardID),
|
||
zap.Error(err))
|
||
}
|
||
}
|
||
|
||
// OnCardEnabled 卡启用轮询时的回调
|
||
func (c *APICallback) OnCardEnabled(ctx context.Context, cardID uint) {
|
||
c.logger.Debug("API 回调：卡启用轮询", zap.Uint("card_id", cardID))
|
||
// 启用后，scheduler 下次扫描时会将其加入队列
|
||
}
|
||
|
||
// OnCardDisabled 卡禁用轮询时的回调
|
||
// 从所有队列中移除卡
|
||
func (c *APICallback) OnCardDisabled(ctx context.Context, cardID uint) {
|
||
c.logger.Debug("API 回调：卡禁用轮询", zap.Uint("card_id", cardID))
|
||
|
||
member := strconv.FormatUint(uint64(cardID), 10)
|
||
|
||
// 从所有轮询队列中移除
|
||
queues := []string{
|
||
constants.RedisPollingQueueRealnameKey(),
|
||
constants.RedisPollingQueueCarddataKey(),
|
||
constants.RedisPollingQueuePackageKey(),
|
||
}
|
||
|
||
for _, queueKey := range queues {
|
||
if err := c.redis.ZRem(ctx, queueKey, member).Err(); err != nil {
|
||
c.logger.Warn("从队列移除卡失败",
|
||
zap.String("queue", queueKey),
|
||
zap.Uint("card_id", cardID),
|
||
zap.Error(err))
|
||
}
|
||
}
|
||
}
|