在现代分布式系统中，消息队列扮演着至关重要的角色：
Apache Kafka 作为分布式消息系统的代表，以其高吞吐、低延迟、高可用的特性，成为大数据领域的首选方案。
+------------------+ +------------------+ +------------------+ | Producer | | Kafka Cluster | | Consumer | | | | | | | | 发送消息到 Topic |---->| Topic 分区存储 |---->| 从 Topic 拉取 | +------------------+ +------------------+ +------------------+ | v +-------------+ | Zookeeper | | 集群协调 | +-------------+
Kafka 采用 顺序写 + 零拷贝 技术：
Topic: orders ├── Partition 0: [msg1][msg2][msg3]... (顺序追加写入) ├── Partition 1: [msg4][msg5][msg6]... └── Partition 2: [msg7][msg8][msg9]... 每个 Partition 是一个有序的日志文件 Consumer 通过 Offset 标记消费位置
gopackage main
import (
"context"
"log"
"github.com/IBM/sarama"
)
type Order struct {
OrderID string `json:"order_id"`
UserID string `json:"user_id"`
Amount float64 `json:"amount"`
}
func main() {
config := sarama.NewConfig()
config.Producer.RequiredAcks = sarama.WaitForAll // 等待所有副本确认
config.Producer.Retry.Max = 5 // 重试次数
config.Producer.Return.Successes = true // 返回成功消息
producer, err := sarama.NewSyncProducer([]string{"localhost:9092"}, config)
if err != nil {
log.Fatal("NewSyncProducer err:", err)
}
defer producer.Close()
// 发送消息
msg := &sarama.ProducerMessage{
Topic: "orders",
Value: sarama.StringEncoder(`{"order_id":"123","user_id":"456","amount":99.9}`),
}
partition, offset, err := producer.SendMessage(msg)
if err != nil {
log.Println("Send err:", err)
return
}
log.Printf("发送成功: partition=%d, offset=%d", partition, offset)
}
gofunc asyncProducer() {
config := sarama.NewConfig()
config.Producer.Flush.Bytes = 1024 * 1024 // 1MB 批量发送
config.Producer.Flush.Messages = 100 // 100 条批量发送
config.Producer.Flush.Frequency = 100 * time.Millisecond // 100ms 刷新
producer, _ := sarama.NewAsyncProducer([]string{"localhost:9092"}, config)
defer producer.Close()
// 处理发送结果
go func() {
for {
select {
case success := <-producer.Successes():
log.Printf("成功: %v", success)
case err := <-producer.Errors():
log.Printf("失败: %v", err)
}
}
}()
// 异步发送
for i := 0; i < 10000; i++ {
msg := &sarama.ProducerMessage{
Topic: "orders",
Value: sarama.StringEncoder(fmt.Sprintf("order-%d", i)),
}
producer.Input() <- msg
}
}
go// 自定义分区器：保证相同 key 的消息进入同一分区
type OrderPartitioner struct{}
func (p *OrderPartitioner) Partition(msg *sarama.ProducerMessage, numPartitions int32) (int32, error) {
// 根据 OrderID 哈希分区
key, _ := msg.Key.Encode()
hash := fnv.New32a()
hash.Write(key)
return int32(hash.Sum32()) % numPartitions, nil
}
func (p *OrderPartitioner) RequiresConsistency() bool {
return true // 保证相同 key 的消息顺序性
}
// 使用自定义分区器
config.Producer.Partitioner = &OrderPartitioner{}
gofunc consumerGroup() {
config := sarama.NewConfig()
config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRoundRobin
config.Consumer.Offsets.Initial = sarama.OffsetNewest // 从最新消息开始
group, _ := sarama.NewConsumerGroup(
[]string{"localhost:9092"},
"order-processors", // 消费者组名
config,
)
defer group.Close()
handler := &OrderHandler{}
for {
// 持续消费
err := group.Consume(context.Background(), []string{"orders"}, handler)
if err != nil {
log.Println("Consume err:", err)
}
}
}
type OrderHandler struct{}
func (h *OrderHandler) Setup(sarama.ConsumerGroupSession) error { return nil }
func (h *OrderHandler) Cleanup(sarama.ConsumerGroupSession) error { return nil }
func (h *OrderHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
for msg := range claim.Messages() {
// 处理消息
log.Printf("处理消息: %s", string(msg.Value))
// 标记消息已处理
session.MarkMessage(msg, "")
}
return nil
}
gofunc (h *OrderHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
for msg := range claim.Messages() {
// 处理消息
err := processOrder(msg.Value)
if err != nil {
log.Printf("处理失败: %v, 稍后重试", err)
continue // 不提交 offset，下次重新消费
}
// 手动提交 offset
session.MarkMessage(msg, "")
session.Commit() // 立即提交
}
return nil
}
go// 不同级别的可靠性保证
config.Producer.RequiredAcks = sarama.NoResponse // 不等待确认（最快，可能丢失）
config.Producer.RequiredAcks = sarama.WaitForLocal // 等待 Leader 确认
config.Producer.RequiredAcks = sarama.WaitForAll // 等待所有 ISR 副本确认（最可靠）
go// 实现精确一次消费
type ExactlyOnceProcessor struct {
db *sql.DB
}
func (p *ExactlyOnceProcessor) Process(msg *sarama.ConsumerMessage) error {
tx, _ := p.db.Begin()
defer tx.Rollback()
// 1. 检查是否已处理（幂等性）
var count int
tx.QueryRow("SELECT COUNT(*) FROM processed_msgs WHERE msg_key = ?", msg.Key).Scan(&count)
if count > 0 {
return nil // 已处理，跳过
}
// 2. 业务处理
_, err := tx.Exec("INSERT INTO orders ...")
if err != nil {
return err
}
// 3. 记录已处理
_, err = tx.Exec("INSERT INTO processed_msgs (msg_key, offset) VALUES (?, ?)", msg.Key, msg.Offset)
if err != nil {
return err
}
return tx.Commit()
}
go// 启用压缩，减少网络传输和存储空间
config.Producer.Compression = sarama.CompressionSnappy // 或 CompressionGZIP
config.Producer.CompressionLevel = 6 // 压缩级别
go// 实现延迟消息
type DelayMessage struct {
Topic string
Payload string
DelayTime time.Time
}
func sendDelayMessage(producer sarama.SyncProducer, topic string, payload string, delay time.Duration) error {
// 使用特定 Topic 存储延迟消息
delayTopic := fmt.Sprintf("delay_%d", time.Now().Add(delay).Unix())
msg := &sarama.ProducerMessage{
Topic: delayTopic,
Value: sarama.StringEncoder(payload),
}
return producer.SendMessage(msg)
}
// 延迟消息消费者
func delayConsumer() {
// 定期扫描延迟 Topic，到期后转发到目标 Topic
ticker := time.NewTicker(1 * time.Second)
for range ticker.C {
// 检查是否有到期的延迟 Topic
// 转发消息到实际 Topic
}
}
go// 批量发送配置
config.Producer.Flush.Bytes = 1024 * 1024 // 批量大小 1MB
config.Producer.Flush.Messages = 1000 // 批量消息数
config.Producer.Flush.Frequency = 10 * time.Millisecond // 刷新间隔
// 并发发送
config.Producer.MaxMessageBytes = 1024 * 1024 // 单条消息最大 1MB
config.Producer.Return.Successes = true
go// 提高吞吐量
config.Consumer.Fetch.Min = 1 // 最小拉取字节数
config.Consumer.Fetch.Max = 1024 * 1024 // 最大拉取字节数 1MB
config.Consumer.Fetch.Default = 1024 * 512 // 默认拉取 512KB
config.Consumer.MaxProcessingTime = 100 * time.Millisecond // 最大处理时间
config.Consumer.Group.Session.Timeout = 10 * time.Second // 会话超时
bash# server.properties
# 网络线程数
num.network.threads=8
num.io.threads=8
# 日志配置
log.flush.interval.messages=10000
log.flush.interval.ms=1000
log.retention.hours=168 # 保留 7 天
# 副本配置
default.replication.factor=3
min.insync.replicas=2
go// 使用 Prometheus 监控
type KafkaMetrics struct {
// 生产者指标
MessagesProduced prometheus.Counter
ProduceLatency prometheus.Histogram
ProduceErrors prometheus.Counter
// 消费者指标
MessagesConsumed prometheus.Counter
ConsumeLatency prometheus.Histogram
ConsumerLag prometheus.Gauge // 消费延迟
// Broker 指标
PartitionCount prometheus.Gauge
UnderReplicated prometheus.Gauge
OfflinePartitions prometheus.Gauge
}
gofunc monitorLag(brokers []string, topic string, groupID string) {
client, _ := sarama.NewClient(brokers, nil)
defer client.Close()
// 获取 Topic 分区信息
partitions, _ := client.Partitions(topic)
for _, partition := range partitions {
// 最新 offset
newestOffset, _ := client.GetOffset(topic, partition, sarama.OffsetNewest)
// 消费者组 offset
offsetManager, _ := sarama.NewOffsetManagerFromClient(groupID, client)
defer offsetManager.Close()
partitionOffsetManager, _ := offsetManager.ManagePartition(topic, partition)
defer partitionOffsetManager.Close()
consumedOffset, _ := partitionOffsetManager.NextOffset()
// 计算延迟
lag := newestOffset - consumedOffset
log.Printf("Partition %d: lag=%d", partition, lag)
}
}
Kafka 作为高吞吐分布式消息系统，核心优势在于：
在实际应用中，需要根据业务场景合理配置生产者、消费者和 Broker 参数，并做好监控和运维工作，才能充分发挥 Kafka 的性能优势。
本文作者:yowayimono
本文链接:
版权声明:本博客所有文章除特别声明外，均采用 BY-NC-SA 许可协议。转载请注明出处！