go *** 作kafka，环境搭建

环境搭建

ps：服务器的kafka环境看上篇文章，此模块记录的是windows/linux go的环境搭建~

1.下载 v.1.19.0 版本的 github.com/Shopify/sarama 模块（windows，linux随意版本）

go get github.com/Shopify/sarama@v.1.19.0

2.kafka/produce/go.mod:

module kafka/produce

go 1.18

require (
	github.com/Shopify/sarama v1.19.0 // indirect
	github.com/davecgh/go-spew v1.1.1 // indirect
	github.com/eapache/go-resiliency v1.2.0 // indirect
	github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21 // indirect
	github.com/eapache/queue v1.1.0 // indirect
	github.com/golang/snappy v0.0.4 // indirect
	github.com/pierrec/lz4 v2.6.1+incompatible // indirect
	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
)

3.kafka/consumer/go.mod:

module kafka/consumer

go 1.18

require github.com/Shopify/sarama v1.19.0

require (
	github.com/davecgh/go-spew v1.1.1 // indirect
	github.com/eapache/go-resiliency v1.2.0 // indirect
	github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21 // indirect
	github.com/eapache/queue v1.1.0 // indirect
	github.com/golang/snappy v0.0.4 // indirect
	github.com/pierrec/lz4 v2.6.1+incompatible // indirect
	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
)

4.file -> setting -> go -> go modules -> Enable Go modules intergration √

代码示例

生产者端（kafka/produce/produce.go）:

package main

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

func main() {
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll          // 发送完数据需要leader和follow都确认
	config.Producer.Partitioner = sarama.NewRandomPartitioner // 新选出一个partition
	config.Producer.Return.Successes = true                   // 成功交付的消息将在success channel返回

	// 连接kafka
	client, err := sarama.NewSyncProducer([]string{"ip:9093"}, config)
	if err != nil {
		fmt.Println("producer closed, err:", err)
		return
	}
	defer func(client sarama.SyncProducer) {
		err := client.Close()
		if err != nil {
			fmt.Println(err)
		}
	}(client)
	var s string
	// 发送消息
	for s != "exit" {
        //从控制台输入字符
		_, err := fmt.Scanln(&s)
		//fmt.Println(s)
		if err != nil {
			fmt.Println(err)
		}
		// 构造一个消息
		msg := &sarama.ProducerMessage{}
		msg.Topic = "web_log"
		msg.Value = sarama.StringEncoder(s)
		pid, offset, err := client.SendMessage(msg)
		if err != nil {
			fmt.Println("send msg failed, err:", err)
		}
		fmt.Printf("pid:%v offset:%v\n", pid, offset)
	}
}

消费者端（kafka/consumer/consumer.go）:

package main

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

func main() {
	consumer, err := sarama.NewConsumer([]string{"ip:9093"}, nil)
	if err != nil {
		fmt.Printf("fail to start consumer, err:%v\n", err)
		return
	}
	partitionList, err := consumer.Partitions("web_log") // 根据topic取到所有的分区
	if err != nil {
		fmt.Printf("fail to get list of partition:err%v\n", err)
		return
	}
	fmt.Println("topic.partitionList:", partitionList)
	// 声明同步锁，保证主线程在任务处理结束前不退出
	var wg sync.WaitGroup
	for partition := range partitionList { // 遍历所有的分区
		// 针对每个分区创建一个对应的分区消费者
		pc, err := consumer.ConsumePartition("web_log", int32(partition), sarama.OffsetNewest)
		if err != nil {
			fmt.Printf("failed to start consumer for partition %d,err:%v\n", partition, err)
			return
		}
		defer func(sarama.PartitionConsumer) {
			pc.AsyncClose()
			wg.Done()
		}(pc)
		// 异步从每个分区消费信息
		wg.Add(1)
		go func(sarama.PartitionConsumer) {
			for msg := range pc.Messages() {
				fmt.Printf("Partition:%d Offset:%d Key:%v Value:%v\n", msg.Partition, msg.Offset, msg.Key, string(msg.Value))
			}
		}(pc)
	}
	wg.Wait()
}