Go语言实现Kafka消息队列的示例代码
目录
- 一、Go语言快速实现Kafka消息队列
- 二、Kafka与其他消息队列对比
- 三、核心差异说明
- 四、选型建议
- 五、常见问题
一、Go语言快速实现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("连接失败php:", 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与其他消息队列对比
对比维度表
特性 | 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), // 相同Keyjs进入同一分区 Value: sarama.StringEncoder("下单成功"), }
如何避免重复消费?
- 启用幂等生产者
config.Producer.Idempotent = true config.Producer.RequiredAcks = sarama.WaitForAll
到此这篇关于Go语言实现Kafka消息队列的示例代码的文章就介绍到这了,更多相关Go Kafka消息队列内容请搜索编程客栈(www.cppcns.cohttp://www.devze.comm)以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程客栈(www.devze.com)!
精彩评论