应根据任务特征选择消息队列：轻量异步用asynq+redis，高吞吐实时流选kafka，混合场景选pulsar；部署需匹配组件特性与运维能力，避免盲目套用。
如何在Kubernetes中选择合适的消息队列组件
别一上来就选Kafka或Pulsar——先看你的任务特征。如果只是轻量级异步任务（比如发邮件、生成报表），Asynq + Redis 就够用，部署快、资源开销小；如果是高吞吐实时流（如IoT设备数据）、需多订阅/多分区语义，Kafka更稳；而Pulsar适合需要统一消息+事件流+函数计算的混合场景，但对ZooKeeper/BookKeeper运维有要求。
常见错误是把Kafka当“万能队列”硬塞进小规模项目：它默认启用副本和持久化，单节点集群容易因insufficient-replicas卡住，且kubectl get pods里kafka-0反复Pending，其实是StorageClass没配好或PV不足。
- Asynq：直接用
helm install asynq . --set redis.host=redis-svc，无需额外协调服务 - Kafka：必须确认
strimzi-kafka-operator已安装，且KafkaCRD版本与Operator匹配（v0.38+对应Kafka 3.6+） - Pulsar：Helm部署时若跳过
--set initialize=true，pulsar-broker会因ZooKeeper未初始化一直CrashLoopBackOff
Go消费者Pod连接Kubernetes内消息队列的坑
本地go run main.go连得通，一打包进镜像就报dial tcp: lookup kafka.default.svc on 10.96.0.10:53: no such host——这是DNS解析失败，不是Kafka没起来。
根本原因：Go默认使用cgo resolver，而Alpine基础镜像没装libc，导致net.LookupHost静默失败。解决方案只有两个：要么换gcr.io/distroless/static镜像（推荐），要么在Dockerfile里加CGO_ENABLED=0编译参数。
- Service DNS名必须写全：
kafka.default.svc.cluster.local，不能只写kafka（尤其跨命名空间时） - 用
sarama.NewConsumerGroup时，config.Consumer.Group.Rebalance.GroupStrategies必须显式设为[]sarama.BalanceStrategy{sarama.NewBalanceStrategyRange()}或RoundRobin，否则消费者组无法分配分区 - 别在init container里跑
kafka-topics.sh创建topic——它依赖JVM，Alpine镜像没Java，改用client-go调Kafka Operator API或用kubectl exec临时Pod
Go协程池消费Redis Streams时的资源泄漏点
用redis.XReadGroup拉取任务时，如果消费者进程意外退出（如OOM Kill），Redis里的consumer信息不会自动清理，导致新Pod启动后XREADGROUP永远返回空——因为旧consumer占着pending entries不ack。
必须在Go程序退出前主动清理：client.XGroupDelConsumer(ctx, "task-stream", "task-group", "consumer-0").Result()。但更稳妥的做法是结合context.WithTimeout和defer：
func startWorker(ctx context.Context, client *redis.Client) {
defer func() {
if r := recover(); r != nil {
client.XGroupDelConsumer(ctx, "task-stream", "task-group", "consumer-"+strconv.Itoa(os.Getpid())).Result()
}
}()
for {
entries, err := client.XReadGroup(ctx, &redis.XReadGroupArgs{
Group: "task-group",
Consumer: "consumer-" + strconv.Itoa(os.Getpid()),
Streams: []string{"task-stream", ">"},
Count: 1,
Block: 5 * time.Second,
}).Result()
if err != nil {
if errors.Is(err, context.DeadlineExceeded) {
continue
}
log.Printf("XReadGroup error: %v", err)
time.Sleep(time.Second)
continue
}
// process...
}
}
- 每个worker用唯一consumer name（比如PID），避免多个Pod共用同一consumer导致消息重复消费
-
XACK必须在任务处理成功后立即执行，否则pending list堆积会拖慢整个stream - 别用
time.Now().Unix()做consumer name——Pod重启后时间戳不变，Redis认为是同一consumer，不分配新消息
异步任务状态同步到Kubernetes资源对象的实操要点
Operator里想把任务执行结果写回CR的状态字段，但r.Status().Update()总报the object has been modified; please apply your changes to the latest version and try again——这不是并发冲突，而是你没在Reconcile里重新Get一遍对象。
标准流程必须是：Get → 修改Status → Update，中间不能穿插其他API调用。如果要等外部系统（比如Kafka任务完成），用ctrl.Result{RequeueAfter: 30 * time.Second}轮询，而不是阻塞goroutine。
- 状态字段设计要区分phase和condition：
Status.Phase表示宏观阶段（Running/Succeeded），Status.Conditions存具体事件（TaskCompleted、RetriesExhausted） - 更新Status时用
r.Status().Patch(ctx, obj, client.MergeFrom(oldObj))比Update更安全，避免覆盖他人修改 - 别在status里存大字段（比如完整日志），Kubernetes etcd单key限制1MB，超限直接拒绝写入
最易被忽略的是：消费者Pod重启后，Redis Streams里的pending消息没人处理，除非你在consumer启动时主动XCLAIM超时任务，否则这些任务就永远卡住——这和Kafka的rebalance机制完全不同，得自己补。
