golang连接kafka的示例代码

./bin/windows/zookeeper-server-start.bat config/zookeeper.properties

./bin/windows/kafka-server-start.bat config/server.properties

package main

import (
	"fmt"
	"time"

	"github.com/Shopify/sarama"
)

func main() {
	config:=sarama.NewConfig()
	// 生产者配置
	config.Producer.RequiredAcks=sarama.WaitForAll
	config.Producer.Partitioner=sarama.NewRandomPartitioner
	config.Producer.Return.Successes=true
	// 封装消息
	msg:=&sarama.ProducerMessage{}
	msg.Topic="shopping"
	time_str:=time.Now().Format("2006-01-02 15:04:05")
	msg.Value=sarama.StringEncoder("0413 test log!"+time_str)
	// 连接kafka
	client,err:=sarama.NewSyncProducer([]string{"127.0.0.1:9092"}, config)
	if err!=nil {
		fmt.Println("producer closed", err)
		return
	}
	defer client.Close()
	// 发送消息
	partition,offset,err:=client.SendMessage(msg)
	if err!=nil {
		fmt.Println("send failed", err)
		return
	}
	fmt.Printf("partition:%v offset:%v", partition, offset)
}

bin/windows/kafka-console-consumer.bat --bootstrap-server 127.0.0.1:9092 --topic shopping --from-beginning