Go 客户端 franz-go

go get github.com/twmb/franz-go/pkg/kgo    # 核心客户端（生产/消费/事务）
go get github.com/twmb/franz-go/pkg/kadm   # 管理与监控：建 Topic、看组、算 Lag

cl, err := kgo.NewClient(
    kgo.SeedBrokers("localhost:9092", "localhost:9093", "localhost:9094"),
    kgo.ClientID("order-svc"), // 对应 Java 的 client.id，会出现在 Broker 日志/配额里
)
if err != nil {
    log.Fatal(err)
}
defer cl.Close() // 会 flush 在途批次——生产端退出前必须调

if err := cl.Ping(ctx); err != nil {
    log.Fatal("集群不可达: ", err) // 启动自检好去处
}

import "github.com/twmb/franz-go/pkg/sasl/scram"

cl, _ := kgo.NewClient(
    kgo.SeedBrokers("kafka.internal:9093"),
    kgo.DialTLSConfig(&tls.Config{ServerName: "kafka.internal"}), // 用系统 CA 可简写 kgo.DialTLS()
    kgo.SASL(scram.Auth{User: "svc-order", Pass: "s3cret"}.AsSha512Mechanism()),
)

rec := &kgo.Record{
    Topic:   "order-events",
    Key:     []byte("SO20261013001"), // 路由键；nil 即"无键"
    Value:   payload,                 // 序列化自理，库不提供 Serializer 抽象（同 kafka-go）
    Headers: []kgo.RecordHeader{{Key: "x-msg-id", Value: []byte(msgID)}},
}

res := cl.ProduceSync(ctx, rec)
if err := res.FirstErr(); err != nil {
    // 分类处理见"错误分类"小节；超时前 franz 内部已按可重试性自动重试
}
pr := res[0] // ProduceResults 与入参逐条对应
log.Printf("写入 topic=%s partition=%d offset=%d",
    pr.Record.Topic, pr.Record.Partition, pr.Record.Offset)

cl.Produce(ctx, rec, func(r *kgo.Record, err error) {
    if err != nil {
        // 唯一的结果回收口：告警/落兜底表；回调里不要 panic
    }
})
// 在途记录入队后立即返回；Close() 会等在途批次与回调全部完成

cl, _ := kgo.NewClient(
    kgo.SeedBrokers("localhost:9092"),
    kgo.DefaultProduceTopic("order-events"), // 只给"没写 Topic"的 Record 兜底 Topic
    kgo.ProducerBatchCompression(kgo.ZstdCompression(), kgo.SnappyCompression()), // 按偏好传多个，与 Broker 协商
    kgo.ProducerLinger(50*time.Millisecond), // 多攒一点换吞吐/压缩率，延迟敏感就调小
    kgo.MaxBufferedRecords(100_000),         // 内存里待发的记录数上限（默认 5 万）
    kgo.RetryTimeout(10 * time.Second),
)

group, err := kgo.NewClient(
    kgo.SeedBrokers("localhost:9092", "localhost:9093", "localhost:9094"),
    kgo.ConsumerGroup("order-processors"),
    kgo.ConsumeTopics("order-events"),
)
if err != nil {
    log.Fatal(err)
}
defer group.Close()

for {
    fs := group.PollFetches(ctx)
    fs.EachError(func(topic string, partition int32, err error) {
        log.Printf("fetch 错误 %s-%d: %v", topic, partition, err) // 分区级错误不中断整体消费
    })
    fs.EachRecord(func(r *kgo.Record) {
        handle(r)
    })
}

// 全手动提交的标准循环（08 章的幂等消费接着这个骨架讲）
group, _ := kgo.NewClient(
    kgo.SeedBrokers("localhost:9092"),
    kgo.ConsumerGroup("order-processors"),
    kgo.ConsumeTopics("order-events"),
    kgo.DisableAutoCommit(),
)
for {
    fs := group.PollRecords(ctx, 500)
    if err := fs.Err0(); errors.Is(err, context.Canceled) || errors.Is(err, kgo.ErrClientClosed) {
        return // 退出路径：ctx 取消或客户端已 Close，先停 Poll 再收尾
    }
    var done []*kgo.Record
    fs.EachRecord(func(r *kgo.Record) {
        if err := handle(r); err != nil {
            return // 不标记不提交：这条留给后续重放（重试阶梯见 10 章）
        }
        done = append(done, r)
    })
    if len(done) > 0 {
        if err := group.CommitRecords(ctx, done...); err != nil {
            log.Printf("提交失败: %v", err) // 没提上去=下次重放，靠幂等兜底（08 章）
        }
    }
}

kgo.ConsumeStartOffset(kgo.NewOffset().AtEnd()),         // 组首次上线、无已提交位点时的起点；默认 AtStart（earliest！）
kgo.ConsumeResetOffset(kgo.RewindOffset(time.Minute)),   // 位点越界/数据丢失时的恢复策略；默认"回退 1 分钟"

group, _ := kgo.NewClient(
    kgo.SeedBrokers("localhost:9092"),
    kgo.ConsumerGroup("order-processors"),
    kgo.ConsumeTopics("order-events"),
    kgo.Balancers(kgo.CooperativeStickyBalancer()), // 默认就这一个；可选 RoundRobin/Range/Sticky
    kgo.InstanceID("order-processors-2"),           // 静态成员：取 StatefulSet Pod 名，重启不触发重平衡
    kgo.SessionTimeout(30 * time.Second),
    kgo.AutoCommitInterval(5 * time.Second),
)

scan, _ := kgo.NewClient(
    kgo.SeedBrokers("localhost:9092"),
    kgo.ConsumePartitions(map[string]map[int32]kgo.Offset{
        "user-profile": {0: kgo.NewOffset().AtStart()}, // 11 章 compact Topic 全量扫就靠它
        "order-events": {
            3: kgo.NewOffset().AfterMilli(time.Now().Add(-time.Hour).UnixMilli()), // 按时间：分区 3 一小时前起
            5: kgo.NewOffset().AtEnd().Relative(-1000), // 按位点：分区 5 倒数 1000 条起
        },
    }),
)
// PollFetches/PollRecords 与组模式相同，只是没有提交概念——进度在你自己手里

adm := kadm.NewClient(group)

// 组 Lag（等价 --describe --group；巡检/告警直接取总量）
lags, err := adm.Lag(ctx, "order-processors")
for _, dl := range lags.Sorted() {
    log.Printf("group=%s 总Lag=%d", dl.Group, dl.Total())
    for _, l := range dl.Lag.Sorted() { // 分区级明细
        log.Printf("  %s-%d lag=%d member=%v", l.Topic, l.Partition, l.Lag, l.Member)
    }
}

// 建 Topic：6 分区、3 副本、带 Topic 级配置（ptr 是自己写的一行取指针小工具）
resp, err := adm.CreateTopics(ctx, 6, 3,
    map[string]*string{"retention.ms": ptr("259200000")}, "order-events")
resp.EachError(func(name string, err error) { log.Printf("%s 创建失败: %v", name, err) })

// 查组的已提交位点（等价 --describe 的 CURRENT-OFFSET 列）
offs, _ := adm.FetchOffsets(ctx, "order-processors")

var ke *kerr.Error
switch {
case errors.As(err, &ke) && ke.Code == kerr.NotEnoughReplicas.Code:
    // ISR 不足（min.insync.replicas 挡写）：等副本恢复，兜底落库
case errors.As(err, &ke) && ke.Code == kerr.MessageTooLarge.Code:
    // 超限：重试一万次也不会成功（kafka-go 里叫 MessageSizeTooLarge，注意名字差异）
case errors.As(err, &ke) && ke.Retriable:
    // 可重试类（LeaderNotAvailable 等）：走到这里说明 franz 内部重试预算已用尽
default:
    // 非协议错误：网络/超时/序列化，或本地检查失败
}

// 低级别：手动挡，直连目标分区 Leader
conn, _ := kafka.DialLeader(ctx, "tcp", "localhost:9092", "order-events", 0)
defer conn.Close()
conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
_, err = conn.WriteMessages(kafka.Message{Key: []byte("K"), Value: []byte("V")})

// 手动建 Topic：连到 Controller 再 CreateTopics（生产建议显式建，04/13 章）
c, _ := kafka.Dial("tcp", "localhost:9092")
ctrl, _ := c.Controller()
cc, _ := kafka.Dial("tcp", net.JoinHostPort(ctrl.Host, strconv.Itoa(ctrl.Port)))
err = cc.CreateTopics(kafka.TopicConfig{Topic: "order-events", NumPartitions: 6, ReplicationFactor: 3})

r := kafka.NewReader(kafka.ReaderConfig{
    Brokers: []string{"localhost:9092"},
    Topic:   "order-events",
    GroupID: "order-processors",
    MaxBytes: 10e6, // 默认只有 1MB，大消息体务必调大（05 章）
})
defer r.Close() // 组模式下必须优雅断开，否则新消费者接入有延迟

m, err := r.FetchMessage(ctx) // 手动提交模式；ReadMessage 则读一条提一条
if err == nil {
    _ = r.CommitMessages(ctx, m)
}

w := &kafka.Writer{
    Addr:         kafka.TCP("localhost:9092"),
    Topic:        "order-events", // 设了 Writer.Topic 就不能再在 Message 里设 Topic
    RequiredAcks: kafka.RequireAll, // 零值是 RequireNone！
    Balancer:     &kafka.Murmur2Balancer{},
}
err := w.WriteMessages(ctx, kafka.Message{Key: []byte("K"), Value: v})