性能优化

Producer 优化

config := sarama.NewConfig()

// 批量发送
config.Producer.Flush.Bytes = 1024 * 1024  // 1MB
config.Producer.Flush.Messages = 1000
config.Producer.Flush.Frequency = 100 * time.Millisecond

// 压缩
config.Producer.Compression = sarama.CompressionSnappy

// 批次大小
config.Producer.MaxMessageBytes = 1024 * 1024

Consumer 优化

config := sarama.NewConfig()

// 批量消费
config.Consumer.MaxProcessingTime = 100 * time.Millisecond
config.Consumer.Return.Errors = true

// 预取
config.Consumer.Group.Session.Timeout = 10 * time.Second
config.Consumer.Group.Heartbeat.Interval = 3 * time.Second

Broker 优化

# 网络线程
num.network.threads=3

# IO线程
num.io.threads=8

# 堆内存
export KAFKA_HEAP_OPTS="-Xms4g -Xmx4g"

监控指标

• 消息吞吐量
• 消费延迟
• 磁盘使用率
• 网络流量

参考：https://kafka.apache.org/documentation/