Go语言实现Kafka消息队列的示例代码

目录

• 一、Go语言快速实现Kafka消息队列
• 二、Kafka与其他消息队列对比
• 三、核心差异说明
• 四、选型建议
• 五、常见问题

安装依赖库

go get github.com/IBM/sarama

生产者代码示例

package main

import (
	"log"
	"github.com/IBM/sarama"
)

func main() {
	// 1. 创建生产者配置
	config := sarama.NewConfig()
	config.Producer.Return.Successes = true

	// 2. 连接Kafka集群
	producer, err := sarama.NewSyncProducer([]string{"localhost:9092"}, config)
	if err != nil {
		log.Fatal("连接失败:", err)
	}
	defer producer.Close()

	// 3. 发送消息
	msg := &sarama.ProducerMessage{
		Topic: "test-topic",
		Value: sarama.StringEncoder("Hello Kafka!"),
	}
	
	partition, offset, err := producer.SendMessage(msg)
	if err != nil {
		log.Fatal("发送失败:", err)
	}
	
	log.Printf("发送成功! 分区:%d 偏移量:%d", partition, offset)
}

消费者代码示例

package main

import (
	"log"
	"github.com/IBM/sarama"
)

func main() {
	// 1. 创建消费者
	consumer, err := sarama.NewConsumer([]string{"localhost:9092"}, nil)
	if err != nil {
		log.Fatal("连接失败:", err)
	}
	defer consumer.Close()

	// 2. 订阅指定分区（这里消费分区0）
	partitionConsumer, err := consumer.ConsumePartition("test-topic", 0, sarama.OffsetNewest)
	if err != nil {
		log.Fatal("订阅失败:", err)
	}
	defer partitionConsumer.Close()

	// 3. 持续消费消息
	for msg := range partitionConsumer.Messages() {
		log.Printf("收到消息: 分区:%d 偏移量:%d 内容:%s", 
			msg.Partition, msg.Offset, string(msg.Value))
	}
}

对比维度表

特性	Kafka	RabbitMQ	Redis Streams	RocketMQ
设计定位	高吞吐日志处理	企业级消息路由	轻量级实时消息	金融级可靠消息
吞吐量	百万级/秒	万级/秒	十万级/秒	十万级/秒
消息持久化	磁盘持久化（默认7天）	内存/磁盘可选	内存为主	磁盘持久化
消息模式	发布-订阅	点对点/发布订阅	发布订阅	发布订阅
协议支持	自定义协议	AMQP协议	RESP协议	自定义协议
顺序保证	分区内有序	队列有序	全局无序	队列有序
延迟消息	不支持	支持	支持	支持
典型场景	日志收集、流处理	订单系统、任务队列	实时通知、排行榜更新	交易系统、金融场景
Go生态支持	Sarama库完善	amqp库稳定	go-redis库支持	官方SDK支持

吞吐量对比

• Kafka：采用顺序磁盘写入 + 零拷贝技术，适合大数据量场景（如：单集群可达百万TPS）
• RabbitMQ：基于内存的Erlang实现，适合中小规模业务（如：电商订单处理）

消息存储机制

// Kafka的持久化示例（自动创建日志文件）
kafka-topics.sh --create 
--topic user-log 
--partitions 3 
--replication-factor 2 
--config retention.ms=604800000  // 保留7天

• Kafka：消息持久化到磁盘，支持TB级数据存储
• Redis：数据主要在内存，重启可能丢失（需配置持久化）

消息消费模式

• Kafka：消费者主动拉取（Pull）
• RabbitMQ：服务端推送（Push）

使用复杂度

// 对比初始化复杂度（以生产者为例）

// Kafka需要配置更多参数
config := sarama.NewConfig()
config.Producer.Return.Successes = true
config.Producer.Partitioner = sarama.NewHashPartitioner

// RabbitMQ连接更简单
conn, _ := amqp.Dial("amqp://guest:guest@localhost:5672/")

选择Kafka当

• 需要处理日志流、点击流等大数据量场景
• 要求消息持久化存储和回溯能力
• 需要横向扩展的分布式系统

选择其他队列当

• RabbitMQ：需要复杂路由（如：延时队列、死信队列）
• Redis Streams：需要极低延迟的实时通知
• RocketMQ：金融场景要求强一致性

如何保证Kafka消息顺序？

• 单分区内保证顺序
• 发送时指定分区键（如用户ID）

msg := &sarama.ProducerMessage{
    Topic: "order-events",
    Key:   sarama.StringEncoder(userID), // 相同Key进入同一分区
    Value: sarama.StringEncoder("下单成功"),
}

如何避免重复消费？

• 启用幂等生产者

config.Producer.Idempotent = true
config.Producer.RequiredAcks = sarama.WaitForAll

到此这篇关于Go语言实现Kafka消息队列的示例代码的文章就介绍到这了,更多相关Go Kafka消息队列内容请搜索风君子博客以前的文章或继续浏览下面的相关文章希望大家以后多多支持风君子博客！ 

您可能感兴趣的文章:

• Go操作各大消息队列教程(RabbitMQ、Kafka)
• Golang微服务框架Kratos实现Kafka消息队列的方法