代码概述
本模板基于 Kafka + Go + Protobuf，提供企业级的生产者和消费者代码框架，具备以下特性：
- 生产者：连接管理、Protobuf 序列化、可配置分区键、Kafka 发送重试（由 kafka-go 处理）、结构化日志、超时控制。
- 消费者：消费者组消费、Protobuf 反序列化、处理回调、失败重试、DLQ（死信队列）、手动提交偏移量、结构化日志、优雅停机。
- 安全：可选 TLS/SASL（PLAIN/SCRAM），无硬编码敏感信息。
- 配置：支持通过结构体/环境变量/配置文件加载，示例提供 YAML。
适用场景：新项目接入 Kafka、现有系统标准化消息代码、从其他 MQ 迁移到 Kafka 时的代码骨架搭建。
依赖配置
- Go 版本：1.21+
- 依赖模块：
- github.com/segmentio/kafka-go v0.4.x
- google.golang.org/protobuf v1.34.x 及以上
- google.golang.org/protobuf/cmd/protoc-gen-go v1.34.x 及以上（生成 Protobuf 代码）
- 可选（SASL）：
- github.com/segmentio/kafka-go/sasl/plain
- github.com/segmentio/kafka-go/sasl/scram
安装命令：
生产者代码
文件：internal/kafka/producer.go
package mqkafka
import (
"context"
"crypto/tls"
"crypto/x509"
"errors"
"fmt"
"log/slog"
"os"
"time"
"github.com/segmentio/kafka-go"
"github.com/segmentio/kafka-go/sasl"
"github.com/segmentio/kafka-go/sasl/plain"
"github.com/segmentio/kafka-go/sasl/scram"
"google.golang.org/protobuf/proto"
)
// ProducerConfig 定义生产者配置
type ProducerConfig struct {
Brokers []string // Kafka broker 列表，例如: []string{"kafka-1:9092","kafka-2:9092"}
Topic string // 目标主题
ClientID string // 客户端ID，用于标识生产者
Compression kafka.Compression // 压缩算法：kafka.Snappy / kafka.Lz4 / kafka.Gzip / kafka.Zstd / kafka.NoCompression
RequiredAcks kafka.RequiredAcks // 建议使用 kafka.RequireAll 保证可靠性
BatchSize int // 批量大小（消息数）
BatchBytes int64 // 批量字节大小
BatchTimeout time.Duration // 批量发送等待超时
WriteTimeout time.Duration // 写超时
AllowAutoTopicCreate bool // 是否允许自动创建主题（依赖 broker 配置）
SchemaVersion string // 自定义 schema 版本，写入 headers 用于兼容控制
DefaultHeaders map[string]string // 默认消息头（会与每条消息的 headers 合并）
// 安全配置
Security SecurityConfig
}
// SecurityConfig 定义 TLS/SASL 配置
type SecurityConfig struct {
EnableTLS bool
CACertPath string // 可选：CA 证书路径（PEM）
ClientCertPath string // 可选：客户端证书路径（PEM）
ClientKeyPath string // 可选：客户端私钥路径（PEM）
InsecureSkipVerify bool // 不建议在生产使用
EnableSASL bool
SASLMechanism string // 仅当 EnableSASL=true 时有效: "PLAIN" | "SCRAM-SHA-256" | "SCRAM-SHA-512"
SASLUsername string // 从环境/配置加载，勿硬编码
SASLPassword string // 从环境/配置加载，勿硬编码
}
// Producer 生产者实例
type Producer struct {
w *kafka.Writer
topic string
log *slog.Logger
headers map[string]string
}
// NewProducer 构建生产者实例
func NewProducer(cfg ProducerConfig, logger *slog.Logger) (*Producer, error) {
if logger == nil {
logger = slog.Default()
}
if len(cfg.Brokers) == 0 {
return nil, errors.New("missing brokers")
}
if cfg.Topic == "" {
return nil, errors.New("missing topic")
}
if cfg.RequiredAcks == 0 {
cfg.RequiredAcks = kafka.RequireAll
}
if cfg.BatchTimeout == 0 {
cfg.BatchTimeout = 50 * time.Millisecond
}
if cfg.WriteTimeout == 0 {
cfg.WriteTimeout = 5 * time.Second
}
dialer := &kafka.Dialer{
Timeout: 10 * time.Second,
ClientID: cfg.ClientID,
DualStack: true,
}
// TLS
if cfg.Security.EnableTLS {
tlsCfg, err := buildTLSConfig(cfg.Security)
if err != nil {
return nil, fmt.Errorf("build TLS config: %w", err)
}
dialer.TLS = tlsCfg
}
// SASL
if cfg.Security.EnableSASL {
mech, err := buildSASLMechanism(cfg.Security)
if err != nil {
return nil, fmt.Errorf("build SASL mechanism: %w", err)
}
dialer.SASLMechanism = mech
}
w := &kafka.Writer{
Addr: kafka.TCP(cfg.Brokers...),
Topic: cfg.Topic,
Balancer: &kafka.Hash{}, // 使用 key 做哈希选择分区，保证同 key 顺序
Compression: cfg.Compression,
RequiredAcks: cfg.RequiredAcks,
BatchSize: cfg.BatchSize,
BatchBytes: cfg.BatchBytes,
BatchTimeout: cfg.BatchTimeout,
WriteTimeout: cfg.WriteTimeout,
AllowAutoTopicCreation: cfg.AllowAutoTopicCreate,
Transport: &kafka.Transport{
Dial: dialer.DialFunc,
TLS: dialer.TLS, // 由 Dialer 统一处理
},
// Kafka-go 会在内部对暂时性错误进行重试（例如网络抖动、临时不可用），无需手动实现额外重试。
}
defaultHeaders := map[string]string{
"content-type": "application/x-protobuf",
"sdk-language": "go",
"sdk-component": "producer",
}
if cfg.SchemaVersion != "" {
defaultHeaders["schema-version"] = cfg.SchemaVersion
}
// 合并用户默认 headers
for k, v := range cfg.DefaultHeaders {
defaultHeaders[k] = v
}
return &Producer{
w: w,
topic: cfg.Topic,
log: logger.With("component", "kafka_producer", "topic", cfg.Topic),
headers: defaultHeaders,
}, nil
}
// Close 关闭生产者
func (p *Producer) Close() error {
return p.w.Close()
}
// Send 发送 protobuf 消息
// 参数：
// - ctx: 外部上下文（建议带有超时）
// - key: 分区键（相同 key 的消息进入同一分区），可为 nil
// - msg: 任意 protobuf 消息
// - headers: 每条消息的附加 headers（会与默认 headers 合并）
func (p *Producer) Send(ctx context.Context, key []byte, msg proto.Message, headers map[string]string) error {
if msg == nil {
return errors.New("nil protobuf message")
}
// 建议上层设置超时；若未设置，这里设置一个保守的默认超时
if _, ok := ctx.Deadline(); !ok {
var cancel context.CancelFunc
ctx, cancel = context.WithTimeout(ctx, 10*time.Second)
defer cancel()
}
payload, err := proto.Marshal(msg)
if err != nil {
return fmt.Errorf("protobuf marshal: %w", err)
}
allHeaders := mergeHeaders(p.headers, headers)
kmsg := kafka.Message{
Key: key,
Value: payload,
Time: time.Now(),
Headers: toKafkaHeaders(allHeaders),
Topic: p.topic, // 可选，writer 已有 Topic
}
if err := p.w.WriteMessages(ctx, kmsg); err != nil {
p.log.Error("write message failed", "err", err)
return err
}
p.log.Debug("message sent", "bytes", len(payload))
return nil
}
func mergeHeaders(base map[string]string, extra map[string]string) map[string]string {
out := make(map[string]string, len(base)+len(extra))
for k, v := range base {
out[k] = v
}
for k, v := range extra {
out[k] = v
}
return out
}
func toKafkaHeaders(h map[string]string) []kafka.Header {
if len(h) == 0 {
return nil
}
out := make([]kafka.Header, 0, len(h))
for k, v := range h {
out = append(out, kafka.Header{Key: k, Value: []byte(v)})
}
return out
}
func buildTLSConfig(sec SecurityConfig) (*tls.Config, error) {
tlsCfg := &tls.Config{
InsecureSkipVerify: sec.InsecureSkipVerify, // 不建议在生产使用
MinVersion: tls.VersionTLS12,
}
// CA
if sec.CACertPath != "" {
caPEM, err := os.ReadFile(sec.CACertPath)
if err != nil {
return nil, fmt.Errorf("read CA cert: %w", err)
}
cp := x509.NewCertPool()
if !cp.AppendCertsFromPEM(caPEM) {
return nil, errors.New("append CA cert failed")
}
tlsCfg.RootCAs = cp
}
// 客户端证书
if sec.ClientCertPath != "" && sec.ClientKeyPath != "" {
cert, err := tls.LoadX509KeyPair(sec.ClientCertPath, sec.ClientKeyPath)
if err != nil {
return nil, fmt.Errorf("load client cert/key: %w", err)
}
tlsCfg.Certificates = []tls.Certificate{cert}
}
return tlsCfg, nil
}
func buildSASLMechanism(sec SecurityConfig) (sasl.Mechanism, error) {
user := sec.SASLUsername
pass := sec.SASLPassword
if user == "" || pass == "" {
return nil, errors.New("missing SASL credentials")
}
switch sec.SASLMechanism {
case "PLAIN":
return plain.Mechanism{Username: user, Password: pass}, nil
case "SCRAM-SHA-256":
mech, err := scram.Mechanism(scram.SHA256, user, pass)
if err != nil {
return nil, err
}
return mech, nil
case "SCRAM-SHA-512":
mech, err := scram.Mechanism(scram.SHA512, user, pass)
if err != nil {
return nil, err
}
return mech, nil
default:
return nil, fmt.Errorf("unsupported SASL mechanism: %s", sec.SASLMechanism)
}
}
消费者代码
文件：internal/kafka/consumer.go
package mqkafka
import (
"context"
"errors"
"fmt"
"log/slog"
"strconv"
"time"
"github.com/segmentio/kafka-go"
"google.golang.org/protobuf/proto"
)
// Handler 是用户提供的处理回调，业务逻辑应实现此函数。
// 返回 error 表示处理失败，框架将按重试策略处理。
type Handler func(ctx context.Context, key []byte, msg proto.Message, headers map[string]string) error
// NewMessage 是一个工厂方法，用于创建一个空的 protobuf 消息实例，用于反序列化。
// 例如：func() proto.Message { return &pb.Event{} }
type NewMessage func() proto.Message
// ConsumerConfig 定义消费者配置
type ConsumerConfig struct {
Brokers []string
Topic string
GroupID string
ClientID string
MinBytes int // 默认 1
MaxBytes int // 默认 10MB
MaxWait time.Duration // 取消息最大等待
ReadTimeout time.Duration // 读超时
CommitInterval time.Duration // 自动提交间隔；使用手动提交时可设为较长
// 偏移控制（仅在首次无已提交偏移时生效）
StartFromEarliest bool // true: earliest, false: latest
// 处理与重试
OperationTimeout time.Duration // 单条消息处理的超时
MaxRetryAttempts int // 处理失败最大重试次数（不含首次处理），例如 3
RetryBackoff time.Duration // 重试回退时间，默认 500ms
// DLQ
EnableDLQ bool
DLQTopic string
// 安全配置
Security SecurityConfig
}
// Consumer 消费者实例
type Consumer struct {
r *kafka.Reader
dlqWriter *kafka.Writer
log *slog.Logger
handler Handler
newMsg NewMessage
cfg ConsumerConfig
}
// NewConsumer 构建消费者
func NewConsumer(cfg ConsumerConfig, handler Handler, newMsg NewMessage, logger *slog.Logger) (*Consumer, error) {
if logger == nil {
logger = slog.Default()
}
if len(cfg.Brokers) == 0 {
return nil, errors.New("missing brokers")
}
if cfg.Topic == "" {
return nil, errors.New("missing topic")
}
if cfg.GroupID == "" {
return nil, errors.New("missing group id")
}
if handler == nil {
return nil, errors.New("missing handler")
}
if newMsg == nil {
return nil, errors.New("missing message factory")
}
dialer, transport, err := buildDialerTransport(cfg.Security, cfg.ClientID)
if err != nil {
return nil, fmt.Errorf("build dialer: %w", err)
}
if cfg.MinBytes == 0 {
cfg.MinBytes = 1
}
if cfg.MaxBytes == 0 {
cfg.MaxBytes = 10 << 20 // 10MB
}
if cfg.MaxWait == 0 {
cfg.MaxWait = 250 * time.Millisecond
}
if cfg.ReadTimeout == 0 {
cfg.ReadTimeout = 10 * time.Second
}
if cfg.CommitInterval == 0 {
cfg.CommitInterval = time.Second // 实际使用中通常手动提交；该值保守设置
}
if cfg.OperationTimeout == 0 {
cfg.OperationTimeout = 30 * time.Second
}
if cfg.RetryBackoff == 0 {
cfg.RetryBackoff = 500 * time.Millisecond
}
if cfg.MaxRetryAttempts < 0 {
cfg.MaxRetryAttempts = 0
}
readerCfg := kafka.ReaderConfig{
Brokers: cfg.Brokers,
GroupID: cfg.GroupID,
Topic: cfg.Topic,
MinBytes: cfg.MinBytes,
MaxBytes: cfg.MaxBytes,
MaxWait: cfg.MaxWait,
ReadLagInterval: -1, // 关闭默认延迟日志，避免噪声
CommitInterval: cfg.CommitInterval,
StartOffset: func() int64 {
if cfg.StartFromEarliest {
return kafka.FirstOffset
}
return kafka.LastOffset
}(),
Dialer: &kafka.Dialer{
Timeout: dialer.Timeout,
DualStack: dialer.DualStack,
ClientID: cfg.ClientID,
TLS: dialer.TLS,
SASLMechanism: dialer.SASLMechanism,
},
RetentionTime: 0, // 使用 broker 默认
}
r := kafka.NewReader(readerCfg)
r.SetOffsetAt(context.Background(), time.Now()) // 不会改变已有提交偏移，仅保证 reader 完整初始化。可移除。
var dlqWriter *kafka.Writer
if cfg.EnableDLQ {
if cfg.DLQTopic == "" {
return nil, errors.New("dlq enabled but DLQTopic empty")
}
dlqWriter = &kafka.Writer{
Addr: kafka.TCP(cfg.Brokers...),
Topic: cfg.DLQTopic,
Balancer: &kafka.Hash{},
RequiredAcks: kafka.RequireAll,
Transport: transport,
WriteTimeout: 10 * time.Second,
}
}
return &Consumer{
r: r,
dlqWriter: dlqWriter,
log: logger.With("component", "kafka_consumer", "topic", cfg.Topic, "group", cfg.GroupID),
handler: handler,
newMsg: newMsg,
cfg: cfg,
}, nil
}
// Run 启动消费循环（阻塞）。使用 ctx 控制优雅停机。
func (c *Consumer) Run(ctx context.Context) error {
c.log.Info("consumer started")
defer c.log.Info("consumer stopped")
for {
select {
case <-ctx.Done():
return nil
default:
}
// 读取消息（阻塞直到有消息或 ctx/reader 超时）
m, err := c.r.FetchMessage(ctx)
if err != nil {
// 可能是上下文取消或超时，或网络错误
if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
c.log.Warn("fetch context error", "err", err)
continue
}
c.log.Error("fetch message failed", "err", err)
continue
}
// 解析 headers
hdrs := fromKafkaHeaders(m.Headers)
// 反序列化 Protobuf
pm := c.newMsg()
if err := proto.Unmarshal(m.Value, pm); err != nil {
c.log.Error("protobuf unmarshal failed; sending to DLQ (if enabled)", "err", err, "partition", m.Partition, "offset", m.Offset)
// 不可反序列化的消息：直接 DLQ；若 DLQ 不可用则不提交以便后续人工干预
if c.cfg.EnableDLQ {
_ = c.sendToDLQ(ctx, m, hdrs, err, -1) // -1 表示由于解析失败
if commitErr := c.r.CommitMessages(ctx, m); commitErr != nil {
c.log.Error("commit after DLQ failed", "err", commitErr)
}
continue
}
// 无 DLQ：不提交，让消息保留
continue
}
// 处理回调，带超时
processCtx, cancel := context.WithTimeout(ctx, c.cfg.OperationTimeout)
err = c.callWithRetry(processCtx, m, pm, hdrs)
cancel()
if err != nil {
// 超过重试，发送 DLQ 并提交偏移避免毒丸阻塞
if c.cfg.EnableDLQ {
if dlqErr := c.sendToDLQ(ctx, m, hdrs, err, c.cfg.MaxRetryAttempts); dlqErr != nil {
// DLQ 发送失败：不提交偏移，以便稍后重试
c.log.Error("send to DLQ failed; will not commit offset", "err", dlqErr, "partition", m.Partition, "offset", m.Offset)
continue
}
// DLQ 成功 -> 提交偏移
if commitErr := c.r.CommitMessages(ctx, m); commitErr != nil {
c.log.Error("commit after DLQ failed", "err", commitErr)
}
continue
}
// 无 DLQ：不提交，让消息保留以便运维处理
c.log.Error("message permanently failed without DLQ; offset not committed", "err", err, "partition", m.Partition, "offset", m.Offset)
continue
}
// 成功处理 -> 提交偏移
if err := c.r.CommitMessages(ctx, m); err != nil {
c.log.Error("commit message failed", "err", err, "partition", m.Partition, "offset", m.Offset)
continue
}
c.log.Debug("message processed & committed", "partition", m.Partition, "offset", m.Offset)
}
}
// Close 关闭消费者（Reader/DLQ Writer）
func (c *Consumer) Close() error {
var err1, err2 error
if c.r != nil {
err1 = c.r.Close()
}
if c.dlqWriter != nil {
err2 = c.dlqWriter.Close()
}
if err1 != nil {
return err1
}
return err2
}
func (c *Consumer) callWithRetry(ctx context.Context, m kafka.Message, pm proto.Message, headers map[string]string) error {
attempts := 0
for {
select {
case <-ctx.Done():
return ctx.Err()
default:
}
err := c.handler(ctx, m.Key, pm, headers)
if err == nil {
return nil
}
attempts++
c.log.Warn("handler failed", "attempt", attempts, "err", err)
if attempts > c.cfg.MaxRetryAttempts {
return fmt.Errorf("max attempts reached: %w", err)
}
// 回退等待
select {
case <-time.After(c.cfg.RetryBackoff):
case <-ctx.Done():
return ctx.Err()
}
}
}
func (c *Consumer) sendToDLQ(ctx context.Context, m kafka.Message, headers map[string]string, rootErr error, attempts int) error {
if c.dlqWriter == nil {
return errors.New("dlq writer not configured")
}
h := mergeHeaders(headers, map[string]string{
"x-error": truncateErr(rootErr, 2048),
"x-retries": strconv.Itoa(attempts),
"content-type": "application/x-protobuf",
"sdk-component": "dlq",
})
dlqMsg := kafka.Message{
Key: m.Key,
Value: m.Value, // 保留原始 payload
Time: time.Now(),
Headers: toKafkaHeaders(h),
}
if _, ok := ctx.Deadline(); !ok {
var cancel context.CancelFunc
ctx, cancel = context.WithTimeout(ctx, 10*time.Second)
defer cancel()
}
if err := c.dlqWriter.WriteMessages(ctx, dlqMsg); err != nil {
c.log.Error("write dlq message failed", "err", err)
return err
}
c.log.Error("message sent to DLQ", "partition", m.Partition, "offset", m.Offset)
return nil
}
func fromKafkaHeaders(hdrs []kafka.Header) map[string]string {
if len(hdrs) == 0 {
return map[string]string{}
}
m := make(map[string]string, len(hdrs))
for _, h := range hdrs {
m[h.Key] = string(h.Value)
}
return m
}
func truncateErr(err error, n int) string {
if err == nil {
return ""
}
s := err.Error()
if len(s) > n {
return s[:n]
}
return s
}
func buildDialerTransport(sec SecurityConfig, clientID string) (*kafka.Dialer, *kafka.Transport, error) {
dialer := &kafka.Dialer{
Timeout: 10 * time.Second,
ClientID: clientID,
DualStack: true,
}
if sec.EnableTLS {
tlsCfg, err := buildTLSConfig(sec)
if err != nil {
return nil, nil, err
}
dialer.TLS = tlsCfg
}
if sec.EnableSASL {
mech, err := buildSASLMechanism(sec)
if err != nil {
return nil, nil, err
}
dialer.SASLMechanism = mech
}
transport := &kafka.Transport{
Dial: dialer.DialFunc,
TLS: dialer.TLS,
}
return dialer, transport, nil
}
使用示例（集成业务逻辑）：
// 假设已生成 pb 包：package pb; message Event { string id = 1; string type = 2; bytes payload = 3; int64 ts_unix_ms = 4; }
// 生产者发送：
// prod, _ := mqkafka.NewProducer(prodCfg, slog.Default())
// defer prod.Close()
// evt := &pb.Event{Id: "k1", Type: "created", Payload: []byte("..."), TsUnixMs: time.Now().UnixMilli()}
// _ = prod.Send(ctx, []byte(evt.Id), evt, map[string]string{"x-source":"svc-a"})
// 消费者处理：
// handler := func(ctx context.Context, key []byte, msg proto.Message, headers map[string]string) error {
// evt := msg.(*pb.Event)
// // TODO: 业务处理逻辑
// return nil
// }
// cons, _ := mqkafka.NewConsumer(consCfg, handler, func() proto.Message { return &pb.Event{} }, slog.Default())
// defer cons.Close()
// _ = cons.Run(ctx)
配置示例
- YAML 示例（config/kafka.yaml）
producer:
brokers: ["kafka-1:9092","kafka-2:9092"]
topic: "events"
clientID: "svc-a-producer"
compression: "snappy" # 可选: none|snappy|lz4|gzip|zstd
requiredAcks: "all" # 可选: none|leader|all
batchSize: 100
batchBytes: 1048576
batchTimeout: "50ms"
writeTimeout: "5s"
allowAutoTopicCreate: false
schemaVersion: "v1"
defaultHeaders:
x-env: "prod"
security:
enableTLS: true
caCertPath: "/etc/ssl/certs/ca.pem"
clientCertPath: "/etc/ssl/certs/client.pem"
clientKeyPath: "/etc/ssl/private/client.key"
insecureSkipVerify: false
enableSASL: true
saslMechanism: "SCRAM-SHA-512"
saslUsername: "${KAFKA_USERNAME}" # 建议从环境变量加载
saslPassword: "${KAFKA_PASSWORD}"
consumer:
brokers: ["kafka-1:9092","kafka-2:9092"]
topic: "events"
groupID: "svc-b-consumer"
clientID: "svc-b-consumer"
minBytes: 1
maxBytes: 10485760
maxWait: "250ms"
readTimeout: "10s"
commitInterval: "1s"
startFromEarliest: true
operationTimeout: "30s"
maxRetryAttempts: 3
retryBackoff: "500ms"
enableDLQ: true
dlqTopic: "events.DLQ"
security:
enableTLS: true
caCertPath: "/etc/ssl/certs/ca.pem"
clientCertPath: "/etc/ssl/certs/client.pem"
clientKeyPath: "/etc/ssl/private/client.key"
insecureSkipVerify: false
enableSASL: true
saslMechanism: "SCRAM-SHA-512"
saslUsername: "${KAFKA_USERNAME}"
saslPassword: "${KAFKA_PASSWORD}"
- 环境变量示例（仅示意）
- KAFKA_BROKERS=kafka-1:9092,kafka-2:9092
- KAFKA_TOPIC=events
- KAFKA_GROUP_ID=svc-b-consumer
- KAFKA_USERNAME=your-username
- KAFKA_PASSWORD=your-password
Protobuf 消息定义示例
文件：proto/event.proto
syntax = "proto3";
package example.events;
option go_package = "your.module/path/internal/pb;pb";
message Event {
string id = 1; // 用于作为 Kafka 分区 key 的候选值
string type = 2; // 事件类型
bytes payload = 3; // 业务载荷（二进制）
int64 ts_unix_ms = 4; // 事件时间戳（毫秒）
}
生成命令（在项目根目录）：
- protoc --go_out=. proto/event.proto
生成后在 Go 代码中引用：
- import pb "your.module/path/internal/pb"
使用说明
- 关于可靠性语义：
- 生产者默认 acks=all，配合 Kafka ISR 提供较高可靠性。
- 消费者采用“处理成功后手动提交”策略，保证至少一次投递（at-least-once）。如需避免重复处理，请在业务侧实现幂等。
- 关于并发与吞吐：
- 该模板按顺序消费，优先确保正确性。建议通过增加消费者实例数量（同 GroupID）横向扩展分区级并发。
- 如需在单进程内并发处理，请基于分区维度实现有序并发队列，并仅在分区内按顺序提交偏移，避免乱序导致的可见性问题。
- 关于 DLQ：
- 当超出最大重试次数或消息不可反序列化时，消费者将消息原样投递到 DLQ，并带上错误和重试次数头信息。DLQ 发送成功后提交偏移，避免毒丸消息阻塞主流程。
- 安全注意事项：
- 不要在代码中硬编码任何凭据（用户名、密码、证书）。请使用环境变量或密钥管理系统。
- 避免设置 InsecureSkipVerify=true；如必须使用，请仅限非生产环境并确保风险可控。
- 监控与可观测性：
- 建议结合 slog 的 Handler 输出至日志系统（如 Loki/ELK），并在关键处添加指标（Prometheus）以跟踪生产/消费速率、DLQ 数量、延迟等。
- 典型集成步骤：
- 定义并生成 Protobuf 消息（proto/event.proto -> pb）。
- 初始化 Producer/Consumer 配置（可以从 YAML/环境变量加载）。
- 生产者：构造 pb.Event，选择合适 key（如 event.id），调用 Producer.Send。
- 消费者：实现 Handler，将 proto.Message 断言为 *pb.Event，处理完成后返回 nil。
- 在应用启动/退出时，确保调用 Close 进行资源清理。
- 版本与兼容：
- 模板基于 kafka-go v0.4.x API，如后续版本有破坏性变更，请参考官方文档进行调整。
如需进一步定制（例如：自定义分区策略、批处理发送/处理、Schema Registry 集成、压测与参数调优），请告知具体需求。
