Kubernetes环境下Go微服务中NSQD粘性连接的负载均衡实现方案咨询

嗨，这个问题我之前帮不少人解决过——本质上是nsq长连接特性和K8s Service DNS解析的冲突导致的流量倾斜。你的临时方案虽然能凑合用，但频繁重启生产者肯定不是长久之计，给你几个更靠谱的方案，从简单到云原生最优解都有：

因为你的nsqd是用StatefulSet部署的，每个Pod的DNS名称是固定格式的：{statefulset-name}-{pod-index}.{service-name}.{namespace}.svc.cluster.local。你可以直接利用这个特性，创建多个Producer实例，每个连接一个nsqd Pod，然后发送消息时轮询或者随机选择实例。

举个代码例子：

package main

import (
    "fmt"
    "log"
    "sync/atomic"

    "github.com/nsqio/go-nsq"
)

var producers []*nsq.Producer
var currentIdx int32

func initProducers(podCount int, serviceName, namespace string) error {
    config := nsq.NewConfig()
    for i := 0; i < podCount; i++ {
        podAddr := fmt.Sprintf("%s-%d.%s.%s.svc.cluster.local:4150", "nsqd", i, serviceName, namespace)
        p, err := nsq.NewProducer(podAddr, config)
        if err != nil {
            return fmt.Errorf("failed to create producer for %s: %w", podAddr, err)
        }
        producers = append(producers, p)
    }
    return nil
}

func publish(topic string, msg []byte) error {
    // 原子操作保证线程安全的轮询
    idx := atomic.AddInt32(&currentIdx, 1) % int32(len(producers))
    return producers[idx].Publish(topic, msg)
}

func main() {
    // 假设你有3个nsqd Pod，Service名叫nsqd，命名空间是default
    if err := initProducers(3, "nsqd", "default"); err != nil {
        log.Fatal(err)
    }
    // 后续调用publish发送消息即可
}

这个方案的好处是不需要额外组件，缺点是如果StatefulSet的Pod数量变化，你需要手动调整代码或者通过环境变量传入Pod数，动态扩容缩容时可以结合K8s API监听Pod变化来自动维护生产者列表。

这其实是nsq官方设计的标准模式——用nsqlookupd来做服务发现，让生产者自动感知所有可用的nsqd节点，并且自动实现负载均衡。

步骤很简单：

1. 先在K8s里部署nsqlookupd的Deployment和Service（因为nsqlookupd是无状态的，用Deployment就行）。
2. 修改你的生产者代码，连接到nsqlookupd而不是直接连nsqd：

package main

import (
    "log"

    "github.com/nsqio/go-nsq"
)

func main() {
    config := nsq.NewConfig()
    // 这里不需要指定nsqd地址，留空即可
    producer, err := nsq.NewProducer("", config)
    if err != nil {
        log.Fatal(err)
    }

    // 连接到nsqlookupd的Service地址（比如nsqlookupd:4161）
    err = producer.ConnectToNSQLookupd("nsqlookupd:4161")
    if err != nil {
        log.Fatal(err)
    }

    // 发送消息时，producer会自动在所有发现的nsqd节点间做负载均衡
    err = producer.Publish("test_topic", []byte("hello nsq"))
    if err != nil {
        log.Fatal(err)
    }
}

这样做的好处完全适配云原生环境：

• 自动发现nsqd节点的上下线，不用手动维护节点列表
• 生产者会自动在多个nsqd之间做负载均衡，流量均匀分布
• 支持动态扩容缩容nsqd StatefulSet，不需要修改代码

频繁调用p.Stop()再重新初始化，会导致连接频繁断开，不仅浪费资源，还可能因为连接中断丢失未确认的消息，而且每次重新连接也不一定能保证分到不同的Pod（取决于DNS解析的随机性）。

内容的提问来源于stack exchange，提问作者xakepp35