golang使用kafka

深藏阁楼爱情的钟 2022-03-21 07:22 313阅读 0赞

使用客户端：sarama

需要先安装gcc，windows系统可参考：[https://blog.csdn.net/skh2015java/article/details/85075032][https_blog.csdn.net_skh2015java_article_details_85075032]

生产者

package main
    
    import (
    	"log"
    	"os"
    	"time"
    
    	"github.com/Shopify/sarama"
    )
    var (
    	logger = log.New(os.Stderr, "[sarama]", log.LstdFlags)
    )
    
    func main() {
    	sarama.Logger = logger
    
    	config := sarama.NewConfig()
    	config.Producer.RequiredAcks = sarama.WaitForLocal
    	config.Producer.Flush.Frequency = 500 * time.Millisecond
    	config.Producer.Return.Successes = true
    
    
    	producer, err := sarama.NewSyncProducer([]string{"localhost:9092"}, config)
    	if err != nil {
    		logger.Printf("Failed to produce message: %s", err)
    		os.Exit(500)
    	}
    	defer producer.Close()
    
    	for i := 0 ; i< 20 ; i++ {
    		msg := &sarama.ProducerMessage{}
    		msg.Topic = "hello"
    		msg.Partition = int32(-1)
    		msg.Key = sarama.StringEncoder("key")
    		msg.Value = sarama.ByteEncoder("Hello World!")
    
    		partition, offset, err := producer.SendMessage(msg)
    		if err != nil {
    			logger.Println("Failed to produce message: ", err)
    		}
    		logger.Printf("partition=%d, offset=%d\n", partition, offset)
    		time.Sleep(time.Second)
    	}
    
    	select {}
    
    }

消费者：

package main
    
    import (
    	"os"
    	"sync"
    	"log"
    	"fmt"
    
    	"github.com/Shopify/sarama"
    )
    
    var (
    	wg          sync.WaitGroup
    	consumerLog = log.New(os.Stderr, "[sarama]", log.LstdFlags)
    )
    
    func main() {
    	sarama.Logger = consumerLog
    	consumer, err := sarama.NewConsumer([]string{"localhost:9092"}, nil)
    	if err != nil {
    		consumerLog.Printf("Failed to start consumer: %s\n", err)
    	}
    
    	partitionList, err := consumer.Partitions("hello")
    	if err != nil {
    		consumerLog.Println("Failed to get the list of partitions: ", err)
    		return
    	}
    
    	for _,partition := range partitionList {
    		pc, err := consumer.ConsumePartition("hello", int32(partition), sarama.OffsetNewest)
    		if err != nil {
    			consumerLog.Printf("Failed to start consumer for partition %d: %s\n", partition, err)
    			return
    		}
    		defer pc.AsyncClose()
    		wg.Add(1)
    		go func(sarama.PartitionConsumer) {
    			defer wg.Done()
    			for msg := range pc.Messages() {
    				fmt.Printf("Partition:%d, Offset:%d, Key:%s, Value:%s\n", msg.Partition, msg.Offset, string(msg.Key), string(msg.Value))
    			}
    		}(pc)
    	}
    	wg.Wait()
    	consumerLog.Println("Done consuming topic hello")
    	consumer.Close()
    }

[https_blog.csdn.net_skh2015java_article_details_85075032]: https://blog.csdn.net/skh2015java/article/details/85075032