Golang 连接Kafka

Kafka介绍

Kafka是Apache软件基金会开发的一个开源流处理平台，由Java和Scala编写；Kafka是一种高吞吐、分布式、基于订阅发布的消息系统。

Kafka名称解释

Producer：生产者
Consumer：消费者
Topic：消息主题，每一类的消息称之为一个主题
Broker：Kafka以集群的方式运行，可以由一个或多个服务器组成，每个服务器叫做一个broker
Partition：物理概念上的分区，为了提供系统吞吐量，在物理上每个Topic会分为一个或多个Partition

Kafka架构图

一个典型的Kafka集群中包含若干Producer，若干broker（Kafka支持水平扩展，一般broker数量越多，集群吞吐率越高），若干Consumer Group，以及一个Zookeeper集群。

Kafka通过Zookeeper管理集群配置及服务协同，Producer使用push模式将消息发布到broker，Consumer通过监听使用pull模式从broker订阅并消费消息。

图上有个细节需要注意，producer给broker的过程是push，也就是有数据就推送给broker，而consumer给broker的过程是pull，是通过consumer主动去拉数据的，而不是broker把数据主动发送给consumer端的。

Kafka与RabbitMQ比较

Kafka比RabbitMQ性能要高
RabbitMQ比Kafka可靠性要高
因此在金融支付领域使用RabbitMQ居多，而在日志处理、大数据等方面Kafka使用居多。

Kafka安装

第一步下载Kafka：

　　地址 http://kafka.apache.org/downloads

第二步解压Kafka：

　　tar -zxvf kafka.tgz -C /usr/local/kafka

第三步运行Zookeeper：

以后台方式运行 /usr/local/kafka/bin/zookeeper-server-start.sh /usr/local/kafka/config/zookeeper.properties & zookeeper端口 2181

第四步运行Kafka：

以后台方式运行 /usr/local/kafka/bin/kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties kafka端口 9092

Kafka图形管理工具

http://www.kafkatool.com/download.html

Go语言中使用Kafka

Sarama is an MIT-licensed Go client library for Apache Kafka version 0.8 (and later).

安装sarama

　　go get github.com/Shopify/sarama

Producer

package mainimport ("fmt""github.com/Shopify/sarama"
)func main() {// 新建一个arama配置实例config := sarama.NewConfig()// WaitForAll waits for all in-sync replicas to commit before responding.config.Producer.RequiredAcks = sarama.WaitForAll// NewRandomPartitioner returns a Partitioner which chooses a random partition each time.config.Producer.Partitioner = sarama.NewRandomPartitionerconfig.Producer.Return.Successes = true// 新建一个同步生产者client, err := sarama.NewSyncProducer([]string{"172.16.65.210:9092"}, config)if err != nil {fmt.Println("producer close, err:", err)return}defer client.Close()// 定义一个生产消息，包括Topic、消息内容、msg := &sarama.ProducerMessage{}msg.Topic = "revolution"msg.Key = sarama.StringEncoder("miles")msg.Value = sarama.StringEncoder("hello world...")// 发送消息pid, offset, err := client.SendMessage(msg)msg2 := &sarama.ProducerMessage{}msg2.Topic = "revolution"msg2.Key = sarama.StringEncoder("monroe")msg2.Value = sarama.StringEncoder("hello world2...")pid2, offset2, err := client.SendMessage(msg2)if err != nil {fmt.Println("send message failed,", err)return}fmt.Printf("pid:%v offset:%v\n", pid, offset)fmt.Printf("pid2:%v offset2:%v\n", pid2, offset2)
}

Consumer

package mainimport ("sync""github.com/Shopify/sarama""fmt"
)var wg sync.WaitGroupfunc main() {consumer, err := sarama.NewConsumer([]string{"172.16.65.210:9092"}, nil)if err != nil {fmt.Println("consumer connect error:", err)return}fmt.Println("connnect success...")defer consumer.Close()partitions, err := consumer.Partitions("revolution")if err != nil {fmt.Println("geet partitions failed, err:", err)return}for _, p := range partitions {partitionConsumer, err := consumer.ConsumePartition("revolution", p, sarama.OffsetOldest)if err != nil {fmt.Println("partitionConsumer err:", err)continue}wg.Add(1)go func(){for m := range partitionConsumer.Messages() {fmt.Printf("key: %s, text: %s, offset: %d\n", string(m.Key), string(m.Value), m.Offset)}wg.Done()}()}wg.Wait()
}

转载于:https://www.cnblogs.com/vincenshen/p/9824486.html