Kafka 消息

pkg/core/kafka 把 franz-go 关在里面：一个默认 acks=all 的 生产者、一个 at-least-once 的消费循环，以及可选死信队列。业务代码不 import franz-go。

配了才建

kafka:
  brokers: ${KAFKA_BROKERS:localhost:9092}   # 逗号分隔串或 YAML 列表
  client_id: matex
  dlq_topic: my.events.dlq                    # 留空则不启用 DLQ

brokers 留空即禁用：env.Kafka() 返回 ok=false， env.RequireKafka() 则让启动失败。verticle.Run 按这段配置建 一个生产者给所有业务复用，取到后用 env.KafkaCfg（同一份 brokers / client id）去建消费者。

生产者：同步还是异步

p, err := kafka.NewProducer(kafka.Config{Brokers: []string{"localhost:9092"}, ClientID: "svc"})

err = p.SendSync(ctx, topic, key, value)   // 等 broker 确认；不能丢就用它
p.Send(ctx, topic, key, value)             // fire-and-forget；失败只记日志
err = p.SendJSON(ctx, topic, key, obj)     // 同步 + JSON 序列化
方法 语义 失败怎么表现 什么时候用
SendSync 阻塞到 broker 确认 返回包装过的 error，调用者决定怎么办 关键事件、事件即业务事实
Send 立刻返回 只记一条 obs.Error，绝不冒泡给调用者 可丢的通知、日志、埋点
SendJSON 序列化后走 SendSync 序列化失败或发送失败都返回 error 结构化事件

生产者默认 RequiredAcks(kgo.AllISRAcks())，不额外配置就是持久语义。 Close 会 flush 缓冲记录，由 verticle 在关停时调用。

消费循环

消费者注册在 Setup 里，用 env.Block 让它随应用启动、随 ctx 取消停止： 

func (s *kafkaService) Setup(ctx context.Context, env *verticle.Env) error {
	if err := env.DecodeService(&s.cfg); err != nil {
		return err
	}
	p, ok := env.Kafka()
	if !ok {
		return errors.New("kafkademo: the `kafka:` section is required")
	}
	s.p = p

	if s.cfg.Consume {
		cfg, topic, group := env.KafkaCfg, s.cfg.Topic, s.cfg.Group
		env.Block("consumer", func(ctx context.Context) error {
			return kafka.RunConsumer(ctx, cfg, group, []string{topic}, s.handle)
		})
	}
	return nil
}

handler 签名是 func(ctx context.Context, m *kafka.Message) error， Message 已经归一化成 topic / partition / offset / key / value / time。 循环的语义有四点：

每条消息的 panic 被隔离 handler 在一个带 recover 的包装里运行，panic 会连栈一起转成普通 error。 一条消息里的 bug 不会带走整个消费者——也就不会连带带走同组的每个成员。

at-least-once 意味着可能重复。幂等要自己做：用 key 去重，或写库时靠唯一约束 （见 数据库 的约束冲突映射）。重试与退避属于 handler 内部的事。

死信队列

需要给毒消息留证据、能重放时，把 RunConsumer 换成 RunConsumerDLQ： 

env.Block("consumer", func(ctx context.Context) error {
	dlq, _ := env.RequireKafka()   // 业务不必再建一个生产者
	return kafka.RunConsumerDLQ(ctx, env.KafkaCfg, group, []string{topic}, s.handle, dlq)
})

handler 失败时，消息以 kafka.DLQRecord 的 JSON 形态写入 dlq_topic：

type DLQRecord struct {
	SourceTopic     string    `json:"source_topic"`
	SourcePartition int32     `json:"source_partition"`
	SourceOffset    int64     `json:"source_offset"`
	SourceTime      time.Time `json:"source_time"`
	Key             []byte    `json:"key,omitempty"`
	Value           []byte    `json:"value"`
	Error           string    `json:"error"`
}

原 key 保留，所以按 key 分区的 DLQ 消费者看到的顺序与源 topic 一致； value 是未经改动的原始字节，加上源坐标就能重放这一条。 

flowchart TD
    POLL["PollRecords
一批最多 100 条"] --> H{"handle 单条
（panic 已 recover）"}
    H -->|"nil"| OK["继续下一条"]
    H -->|"error"| LG["obs.Error: handle failed
带 topic/partition/offset"]
    LG --> Q{"dlq_topic 非空
且 dlq 非 nil"}
    Q -->|否| CMT["降级：与 RunConsumer 相同
log-and-continue"]
    Q -->|是| PUB["SendSync 到 DLQ
DLQRecord JSON"]
    PUB -->|成功| WARN["obs.Warn: moved to the dead letter queue"]
    PUB -->|失败| DROP["obs.Error: publish failed;
record is dropped"]
    OK --> CMT
    WARN --> CMT
    DROP --> CMT
    CMT["CommitRecords
整批同步提交"] --> NEXT["下一轮 poll"]
    classDef bad fill:#fbecec,stroke:#b23b3b,color:#2b2722
    class LG,DROP bad
一条失败消息的路径：先写 DLQ，再提交。DLQ 自己写不进去时记 error 后照常提交—— 不能为了一条消息把 partition 卡死。

三条行为约定，都是刻意的：

DLQ 自身也是一个 RunConsumer / RunConsumerDLQ：在它的 handler 里告警或重放。

什么时候不用 Kafka

只需要进程内解耦的话，一个带缓冲的 channel 就够了。引入 Kafka 的门槛是：需要 跨实例、可回放、单 partition 内有序的事件流。

相关示例