Golang操作Kafka：Sarama库Producer与Consumer实战

1. 环境准备与依赖说明

在Go项目中操作Kafka，需引入成熟的客户端库。目前广泛使用的是 github.com/Shopkaft/sarama（注：原始仓库已归档，生产环境建议关注其衍生项目如IBM/sarama或Confluent，但此处保留原文示例以便理解基础用法）。在使用前，请确保已通过 go get 安装该库：

go get github.com/Shopify/sarama

本示例假设 Kafka Broker 的地址为 192.20.216.8:9092，且 Topic task 已经存在。

2. 第一步：构造并发送消息 (Producer)

Kafka的生产者负责将数据推送到指定的Topic。为了数据的可靠性，通常需要配置确认机制（Acknowledgement）。以下代码展示了如何建立同步生产者（SyncProducer）并发送一条字符串消息。

第1步：初始化配置

• 目的：设置连接行为，确保数据不丢失。
• 关键参数：
• config.Producer.RequiredAcks = sarama.WaitForAll：等待Leader和所有ISR（In-Sync Replicas）副本确认后才返回，保障数据一致性。
• config.Producer.Return.Successes = true：将发送成功的消息反馈给调用方，便于监控。

第2步：连接与发送

使用 NewSyncProducer 传入Broker列表和配置进行连接。随后构造 sarama.ProducerMessage，指定Topic为 task，内容为JSON或普通文本。发送成功后，函数将返回 pid (Partition ID) 和 offset (偏移量)，这是判断写入是否落盘的重要依据。

package main

import (
	"fmt"
	"github.com/Shopify/sarama"
)

func main() {
	// 1. 配置Producer
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll // 需leader和follower都确认
	config.Producer.Partitioner = sarama.NewRandomPartitioner // 随机选择分区
	config.Producer.Return.Successes = true // 开启成功消息回传

	// 2. 连接Kafka集群
	client, err := sarama.NewSyncProducer([]string{"192.20.216.8:9092"}, config)
	if err != nil {
		fmt.Println("Failed to connect to Kafka: ", err)
		return
	}
	defer client.Close()

	// 3. 构造消息
	msg := &sarama.ProducerMessage{}
	msg.Topic = "task"
	msg.Value = sarama.StringEncoder("Hello Golang from Kafka Producer!")

	// 4. 发送并获取回执
	pid, offset, err := client.SendMessage(msg)
	if err != nil {
		fmt.Println("Send failed: ", err)
		return
	}
	fmt.Printf("Message sent to Partition: %d, Offset: %d\n", pid, offset)
}

3. 第二步：消费消息 (Consumer)

消费者需要订阅特定Topic的各个分区。由于Kafka是并行架构，一个Topic可能被划分为多个Partition。为了充分利用资源，我们通常为每个Partition启动一个独立的Goroutine来处理消息流。

第1步：获取分区列表

通过 consumer.Partitions(topic) 接口查询目标Topic的所有分区ID。这一步确保了后续能覆盖该Topic下的所有数据段。

第2步：启动分区消费者协程

• 入口：遍历上一步获取的 partitionList。
• 动作：调用 consumer.ConsumePartition，起始位置设为 sarama.OffsetNewest（仅消费新消息）。若设为 sarama.OffsetOldest 则从头开始消费。
• 观察状态：在协程内对 pc.Messages() 进行range循环。这是一个阻塞式迭代器，当有新消息到达时，会打印出Partition、Offset以及Key/Value内容。

package main

import (
	"fmt"
	"github.com/Shopify/sarama"
	"sync"
)

func main() {
	var wg sync.WaitGroup
	
	// 1. 创建消费者实例
	consumer, err := sarama.NewConsumer([]string{"192.20.216.8:9092"}, nil)
	if err != nil {
		fmt.Println("Failed to start consumer: ", err)
		return
	}
	defer consumer.Close()

	topic := "task"
	
	// 2. 获取Topic的所有分区
	partitionList, err := consumer.Partitions(topic)
	if err != nil {
		fmt.Println("Failed to get partitions: ", err)
		return
	}

	// 3. 遍历每个分区，启动消费者
	for partition := range partitionList {
		// 针对当前分区创建消费者，从最新消息开始消费
		pc, err := consumer.ConsumePartition(topic, int32(partition), sarama.OffsetNewest)
		if err != nil {
			fmt.Printf("Failed to consume partition %d: %s\n", partition, err)
			continue
		}

		wg.Add(1)
		// 启动Goroutine并发处理消息
		go func(pc sarama.PartitionConsumer) {
			defer pc.AsyncClose()
			defer wg.Done()
			
			// 循环接收消息
			for msg := range pc.Messages() {
				val, _ := stringDecoder(msg.Value)
				key := string(msg.Key)
				fmt.Printf("Partition:%d | Offset:%d | Key:%s | Value:%s\n",
					msg.Partition, msg.Offset, key, val)
			}
		}(pc)
	}
	
	// 等待所有协程结束（实际场景中通常由外部信号控制退出）
	wg.Wait()
}

// 辅助函数：安全解码
func stringDecoder(enc sarama.Encoder) (string, error) {
	return enc.(sarama.StringEncoder).Encode(), nil
}

4. 注意事项与维护

上述示例代码旨在演示API的基本用法，实际生产环境中需注意以下几点：

1. 错误处理：代码中的 if err 应更严谨地记录日志而非直接打印，并使用适当的重试机制。
2. 优雅退出：wg.Wait() 在此仅为示例，实际应用中应监听中断信号（Signal），在收到退出指令后停止循环并关闭Client，防止资源泄漏。
3. 库的更新：由于 github.com/Shopify/sarama 已不再积极维护，建议在新项目中使用其社区维护版本（如 github.com/IBM/sarama），它们兼容原有API但修复了更多潜在Bug。

以上就是Golang使用Sarama操作Kafka进行消息生产和消费的详细内容，更多关于Golang消息队列中间件集成的资料请关注本站其它相关文章！