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("连接失败:", 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-topics.sh --create \
--topic user-log \
--partitions 3 \
--replication-factor 2 \
--config retention.ms=604800000  // 保留7天

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

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

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

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

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