kafka-golisted

# Go Kafka 专家 使用 Go confluent-kafka-go 开发 Kafka 功能：$ARGUMENTS --- ## 1. 客户端管理 ### 创建生产者 ```go func NewKafkaProducer(brokers string) (*kafka.Producer, error) { return kafka.NewProducer(&kafka.ConfigMap{ "bootstrap.servers": brokers, "acks": "all", "enable.idempotence": true, "retries": 3, "linger.ms": 5, "batch.size": 16384, "compression.type": "lz4", "max.in.flight.requests.per.connection": 5, }) } ``` ### 创建消费者 ```go func NewKafkaConsumer(brokers, groupID string) (*kafka.Consumer, error) { return kafka.NewConsumer(&kafka.ConfigMap{ "bootstrap.servers": brokers, "group.id": groupID, "auto.offset.reset": "earliest", "enable.auto.commit": false, // 手动提交 "session.timeout.ms": 30000, "max.poll.interval.ms": 300000, }) } ``` ### 包装器模式 XKit 的 xkafka 使用**两层包装**架构：基础包装器 + 增强包装器（Tracing/DLQ）。 ```go // 基础包装器 - 暴露底层客户端，提供健康检查和统计 type producerWrapper struct { producer *kafka.Producer // 线程安全 mu sync.Mutex // 保护管理操作(Health/Flush/Close) // atomic 统计: MessagesProduced, BytesProduced, Errors, QueueLength } // Producer() 暴露底层生产者用于直接 Produce() 调用 func (w *producerWrapper) Producer() *kafka.Producer func (w *producerWrapper) Health(ctx context.Context) error func (w *producerWrapper) Stats() ProducerStats func (w *producerWrapper) Close() error // Flush + Close ``` **