kafka — 生产者、消费者与死信队列

pkg/core/kafka 包住 franz-go：NewProducer 默认 acks=all（durable），RunConsumer 是一个 at-least-once 的消费循环。本示例把生产者做成两个 HTTP 端点，把消费者注册成 env.Block，随应用一起启动、随 ctx 取消一起停止。

运行

需要真实的 Kafka，先 make dev（起一个单节点 KRaft broker）。 HTTP 默认监听 :8080，配置文件是目录里的 config.yaml， -conf 默认指向它。 

make dev                      # 起 Kafka（KRaft 单节点）
go run ./examples/kafka       # 等价于 make example NAME=kafka
生产者由框架建，你只借用 verticle.Run 按 kafka: 段创建一个生产者并复用 （p, ok := env.Kafka()；一定要有就用 env.RequireKafka()）， 不需要自己 NewProducer，也不需要自己 Close。

它演示了什么

消费者注册

来自 examples/kafka/main.go 的 Setup。 注意它复用生产者用过的同一份 broker 配置（env.KafkaCfg）， 组名与 topic 来自服务段。 

if s.cfg.Consume {
	// The same brokers/client id the producer used.
	cfg, topic, group := env.KafkaCfg, s.cfg.Topic, s.cfg.Group
	env.Block("consumer", func(ctx context.Context) error {
		obs.Info(ctx, "consumer started", "topic", topic, "group", group)
		return kafka.RunConsumer(ctx, cfg, group, []string{topic}, s.handle)
	})
}

同步生产

来自 publishSync。生产者的错误已经被包过一层，直接返回即可； 成功才增加计数。 

if err := s.p.SendSync(ctx, s.cfg.Topic, []byte(e.Key), []byte(e.Value)); err != nil {
	return nil, err // kafka wrapper already wrapped it
}
s.produced.Add(1)
return map[string]any{"topic": s.cfg.Topic, "key": e.Key, "mode": "sync"}, nil

死信队列

本示例注册的是 kafka.RunConsumer，它遇到 handler 错误是 "记日志后照样提交"。pkg/core/kafka 还提供 RunConsumerDLQ：handler 出错时先把消息以 kafka.DLQRecord（源 topic / partition / offset、key、value、error） 写进 cfg.DLQTopic，写完再提交本批。 

flowchart TD
    P["PollRecords(maxPoll=100)"] --> H{"handle(msg)"}
    H -- "err == nil" --> C["提交本批 offset"]
    H -- "err != nil" --> D{"RunConsumerDLQ
DLQTopic 非空?"}
    D -- "是" --> S["dlq.SendSync(DLQRecord)
保留原 key 与源坐标"]
    S --> C
    D -- "否" --> C
    C --> P
    classDef drop fill:#fdf6f0,stroke:#c8641f,color:#2b2722
    classDef ok fill:#eef7f0,stroke:#2f7d4f,color:#22372a
    class C,S ok
    class D drop
死信路径。只有 RunConsumerDLQ 且 DLQTopic 非空时才 写 DLQ；若连写 DLQ 也失败，消息记日志后仍提交——把 partition 永久堵在一个坏掉的 DLQ 上，比丢掉一条已经记过日志的记录更糟。

验证

curl -i -X POST localhost:8080/api/v1/events       -d '{"key":"u1","value":"hello"}'
curl -i -X POST localhost:8080/api/v1/events/async -d '{"key":"u2","value":"fire-and-forget"}'
curl -i localhost:8080/api/v1/stats

两个生产端点都返回 {"topic":"matex.example.events","key":"u1","mode":"sync"} （异步端 mode 为 async）；空 value 回 400 {"code":40002,"msg":"value is required"}。 /api/v1/stats 返回 {"topic":…,"group":"kafkademo-consumers","produced":N,"consumed":M}—— 消费由后台 env.Block 完成，稍等片刻 consumed 就会涨上来。

测试

main_test.go 用 kfake（进程内 broker，kfake.NewCluster） 起一个单 broker 集群并 seed 一个 topic，然后用真实的 kafka.NewProducer + kafka.RunConsumer 跑一遍 生产-消费闭环。它钉住"发出去的消息能被同组消费者收到且 value 一致"， 不需要 Docker、不需要网络。kfake 是唯一允许示例测试 import 的 franz-go 子包（见 .golangci.yml 的 examples-no-drivers）， 因为它只是测试替身。 

go test ./examples/kafka

运行时记得：起始位点是 AtStart，新组会把 topic 里的历史消息全读一遍； 换组名也会从头消费。

相关