diff --git a/internal/infrastructure/nsqbus/nsqbus.go b/internal/infrastructure/nsqbus/nsqbus.go index b5530dff..f4f7feab 100644 --- a/internal/infrastructure/nsqbus/nsqbus.go +++ b/internal/infrastructure/nsqbus/nsqbus.go @@ -4,6 +4,10 @@ package nsqbus import ( "context" + "net" + "net/http" + "net/url" + "strconv" "strings" "sync" "time" @@ -49,6 +53,8 @@ type ConsumerConfig struct { NSQLookupdAddresses []string MsgTimeout time.Duration MaxAttempts uint16 + // LookupdPollInterval 消费者向 nsqlookupd 查询 nsqd 的间隔;0 表示 15 秒。 + LookupdPollInterval time.Duration } // ConsumerManager 按 topic 动态订阅与取消订阅;每个 topic 一个消费者,max_in_flight 控制本进程对该 topic 的并发。 @@ -124,6 +130,12 @@ func (m *ConsumerManager) subscribe(topic string, maxInFlight int) (*nsq.Consume if m.cfg.MaxAttempts > 0 { cfg.MaxAttempts = m.cfg.MaxAttempts } + cfg.LookupdPollInterval = 15 * time.Second + if m.cfg.LookupdPollInterval > 0 { + cfg.LookupdPollInterval = m.cfg.LookupdPollInterval + } + // 先在 nsqd 上创建 topic:否则首次发布前 nsqlookupd 查不到该 topic,消费者要等下一轮轮询才能发现 nsqd。 + m.ensureTopic(topic) consumer, err := nsq.NewConsumer(topic, m.channel, cfg) if err != nil { return nil, err @@ -213,3 +225,32 @@ func (l nsqLogger) Output(_ int, s string) error { } return nil } + +// ensureTopic 通过 nsqd HTTP 接口创建 topic(已存在时幂等);失败只记日志,消费者仍会在后续轮询中发现。 +func (m *ConsumerManager) ensureTopic(topic string) { + httpAddress := nsqdHTTPAddress(m.cfg.NSQDAddress) + if httpAddress == "" { + return + } + client := http.Client{Timeout: 3 * time.Second} + resp, err := client.Post("http://"+httpAddress+"/topic/create?topic="+url.QueryEscape(topic), "text/plain", nil) + if err != nil { + m.logger.Debug("创建 NSQ topic 失败", zap.String("topic", topic), zap.Error(err)) + return + } + _ = resp.Body.Close() +} + +// nsqdHTTPAddress 由 nsqd TCP 地址推导 HTTP 地址(默认端口 4150 → 4151)。 +func nsqdHTTPAddress(tcpAddress string) string { + host, port, err := net.SplitHostPort(strings.TrimSpace(tcpAddress)) + if err != nil || host == "" { + return "" + } + if port == "4150" { + port = "4151" + } else if value, convErr := strconv.Atoi(port); convErr == nil { + port = strconv.Itoa(value + 1) + } + return net.JoinHostPort(host, port) +} diff --git a/internal/task/carrier_card_executor.go b/internal/task/carrier_card_executor.go index b68601bc..2ff210b2 100644 --- a/internal/task/carrier_card_executor.go +++ b/internal/task/carrier_card_executor.go @@ -270,6 +270,10 @@ func (e *CarrierCardExecutor) writeBack(ctx context.Context, card *model.IotCard } set[carrierqueue.FieldDue(checkType)] = strconv.FormatInt(now.Add(delay).UnixMilli(), 10) default: + // 跳过(如所属账户未开实名管控)视同本轮已检查:记最近成功时间,使请求检查的冷却同样生效。 + if outcome.Result == StepSkipped { + set[carrierqueue.FieldOK(checkType)] = strconv.FormatInt(now.UnixMilli(), 10) + } set[carrierqueue.FieldDue(checkType)] = strconv.FormatInt(normalNext.UnixMilli(), 10) } }