西安北郊网站开发网站推广包括

一、kafka集群搭建　　

　　至于kafka是什么我都不多做介绍了，网上写的已经非常详尽了。

 (没安装java环境的需要先安装 yum -y install java-1.8.0-openjdk*)

1. 下载zookeeper  https://zookeeper.apache.org/releases.html

2. 下载kafka http://kafka.apache.org/downloads

3. 启动zookeeper集群（我的示例是3台机器，后面的kafka也一样，这里就以1台代指3台，当然你也可以只开1台）

　　1)配置zookeeper。 修改复制一份 zookeeper-3.4.13/conf/zoo_sample.cfg 改名成zoo.cfg。修改以下几个参数，改成适合自己机器的。

　　dataDir=/home/test/zookeeper/datadataLogDir=/home/test/zookeeper/logserver.1=10.22.1.1:2888:3888server.2=10.22.1.2:2888:3888server.3=10.22.1.3:2888:3888

　　2) 创建myid文件，确定机器编号。分别在3台机器的/home/test/zookeeper/data目录执行分别执行命令 echo 1 > myid(注意ip为10.22.1.2把1改成2，见上面的配置)

　　3) 启动zookeeper集群。分别进入目录zookeeper-3.4.13/bin 执行 sh zkServer.sh start

4. 启动kafka集群

　　1) 配置kafka。进入kafka_2.11-2.2.0/config。复制3份，分别为server1.properties，server2.properties，server3.properties。修改以下几项(注意对应的机器id)

log.dirs和zookeeper.connect 是一样的。broker.id和listeners分别填对应的id和ip

broker.id=1
listeners=PLAINTEXT://10.22.1.1:9092
log.dirs=/home/test/kafka/log
zookeeper.connect=10.22.1.1:2181,10.22.1.2:2181,10.22.1.3:2181

　　2) 启动kafka集群。分别进入kafka_2.11-2.2.0/bin目录,分别执行sh kafka-server-start.sh ../config/server1.properties (第2台用server2.properties配置文件)

　

二、Golang生产者和消费者

　　目前比较流行的golang版的kafka客户端库有两个：

　　1. https://github.com/Shopify/sarama

　　2. https://github.com/confluentinc/confluent-kafka-go

　　至于谁好谁坏自己去分辨，我用的是第1个，star比较多的。

 

1. kafka生产者代码

　　这里有2点要说明：

　　1)  config.Producer.Partitioner = sarama.NewRandomPartitioner，我分partition用的是随机，如果你想稳定分paritition的话可以自定义，还有轮询和hash方式

　　2) 我的topic是走的外部配置，可以根据自己的需求修改

// Package kafka_producer kafka 生产者的包装
package kafka_producerimport ("github.com/Shopify/sarama""strings""sync""time""github.com/alecthomas/log4go"
)// Config 配置
type Config struct {Topic      string `xml:"topic"`Broker     string `xml:"broker"`Frequency  int    `xml:"frequency"`MaxMessage int    `xml:"max_message"`
}type Producer struct {producer sarama.AsyncProducertopic     stringmsgQ      chan *sarama.ProducerMessagewg        sync.WaitGroupcloseChan chan struct{}
}// NewProducer 构造KafkaProducer
func NewProducer(cfg *Config) (*Producer, error) {config := sarama.NewConfig()config.Producer.RequiredAcks = sarama.NoResponse                                  // Only wait for the leader to ackconfig.Producer.Compression = sarama.CompressionSnappy                            // Compress messagesconfig.Producer.Flush.Frequency = time.Duration(cfg.Frequency) * time.Millisecond // Flush batches every 500msconfig.Producer.Partitioner = sarama.NewRandomPartitionerp, err := sarama.NewAsyncProducer(strings.Split(cfg.Broker, ","), config)if err != nil {return nil, err}ret := &Producer{producer:  p,topic:     cfg.Topic,msgQ:      make(chan *sarama.ProducerMessage, cfg.MaxMessage),closeChan: make(chan struct{}),}return ret, nil
}// Run 运行
func (p *Producer) Run() {p.wg.Add(1)go func() {defer p.wg.Done()LOOP:for {select {case m := <-p.msgQ:p.producer.Input() <- mcase err := <-p.producer.Errors():if nil != err && nil != err.Msg {l4g.Error("[producer] err=[%s] topic=[%s] key=[%s] val=[%s]", err.Error(), err.Msg.Topic, err.Msg.Key, err.Msg.Value)}case <-p.closeChan:break LOOP}}}()for hasTask := true; hasTask; {select {case m := <-p.msgQ:p.producer.Input() <- mdefault:hasTask = false}}}// Close 关闭
func (p *Producer) Close() error {close(p.closeChan)l4g.Warn("[producer] is quiting")p.wg.Wait()l4g.Warn("[producer] quit over")return p.producer.Close()
}// Log 发送log
func (p *Producer) Log(key string, val string) {msg := &sarama.ProducerMessage{Topic: p.topic,Key:   sarama.StringEncoder(key),Value: sarama.StringEncoder(val),}select {case p.msgQ <- msg:returndefault:l4g.Error("[producer] err=[msgQ is full] key=[%s] val=[%s]", msg.Key, msg.Value)}
}

 

2. kafka消费者

　　几点说明：

　　1) kafka一定要选用支持集群的版本

　　2) 里面带了创建topic，删除topic，打印topic的工具

　　3) replication是外面配置的

　　4) 开多个consumer需要在创建topic时设置多个partition。官方的示例当开多个consumer的时候会崩溃，我这个版本不会，我给官方提交了一个PR，还不知道有没有采用

 

// Package main Kafka消费者
package mainimport ("context""encoding/xml""flag""fmt""io/ioutil""log""os""os/signal""runtime""strings""syscall""time""github.com/Shopify/sarama""github.com/alecthomas/log4go"
)// Consumer Consumer配置
type ConsumerConfig struct {Topic       []string `xml:"topic"`Broker      string   `xml:"broker"`Partition   int32    `xml:"partition"`Replication int16    `xml:"replication"`Group       string   `xml:"group"`Version     string   `xml:"version"`
}var (configFile = "" // 配置路径initTopic  = falselistTopic  = falsedelTopic   = ""cfg        = &Config{}
)// Config 配置
type Config struct {Consumer ConsumerConfig `xml:"consumer"`
}func init() {flag.StringVar(&configFile, "config", "../config/consumer.xml", "config file ")flag.BoolVar(&initTopic, "init", initTopic, "create topic")flag.BoolVar(&listTopic, "list", listTopic, "list topic")flag.StringVar(&delTopic, "del", delTopic, "delete topic")}func main() {runtime.GOMAXPROCS(runtime.NumCPU())defer func() {time.Sleep(time.Second)log4go.Warn("[main] consumer quit over!")log4go.Global.Close()}()contents, _ := ioutil.ReadFile(configFile)xml.Unmarshal(contents, cfg)// sarama的loggersarama.Logger = log.New(os.Stdout, fmt.Sprintf("[%s]", "consumer"), log.LstdFlags)// 指定kafka版本，一定要支持kafka集群version, err := sarama.ParseKafkaVersion(cfg.Consumer.Version)if err != nil {panic(err)}config := sarama.NewConfig()config.Version = versionconfig.Consumer.Offsets.Initial = sarama.OffsetOldest// 工具if tool(cfg, config) {return}// kafka consumer clientctx, cancel := context.WithCancel(context.Background())client, err := sarama.NewConsumerGroup(strings.Split(cfg.Consumer.Broker, ","), cfg.Consumer.Group, config)if err != nil {panic(err)}consumer := Consumer{}go func() {for {err := client.Consume(ctx, cfg.Consumer.Topic, &consumer)if err != nil {log4go.Error("[main] client.Consume error=[%s]", err.Error())// 5秒后重试time.Sleep(time.Second * 5)}}}()// os signalsigterm := make(chan os.Signal, 1)signal.Notify(sigterm, syscall.SIGINT, syscall.SIGTERM)<-sigtermcancel()err = client.Close()if err != nil {panic(err)}log4go.Info("[main] consumer is quiting")
}func tool(cfg *Config, config *sarama.Config) bool {if initTopic || listTopic || len(delTopic) > 0 {ca, err := sarama.NewClusterAdmin(strings.Split(cfg.Consumer.Broker, ","), config)if nil != err {panic(err)}if len(delTopic) > 0 { // 删除Topicif err := ca.DeleteTopic(delTopic); nil != err {panic(err)}log4go.Info("delete ok topic=[%s]\n", delTopic)} else if initTopic { // 初始化Topicif detail, err := ca.ListTopics(); nil != err {panic(err)} else {for _, v := range cfg.Consumer.Topic {if d, ok := detail[v]; ok {if cfg.Consumer.Partition > d.NumPartitions {if err := ca.CreatePartitions(v, cfg.Consumer.Partition, nil, false); nil != err {panic(err)}log4go.Info("alter topic ok", v, cfg.Consumer.Partition)}} else {if err := ca.CreateTopic(v, &sarama.TopicDetail{NumPartitions: cfg.Consumer.Partition, ReplicationFactor: cfg.Consumer.Replication}, false); nil != err {panic(err)}log4go.Info("create topic ok", v)}}}}// 显示Topic列表if detail, err := ca.ListTopics(); nil != err {log4go.Info("ListTopics error", err)} else {for k := range detail {log4go.Info("[%s] %+v", k, detail[k])}}if err := ca.Close(); nil != err {panic(err)}return true}return false
}type Consumer struct {
}func (consumer *Consumer) Setup(s sarama.ConsumerGroupSession) error {return nil
}func (consumer *Consumer) Cleanup(s sarama.ConsumerGroupSession) error {return nil
}func (consumer *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {for message := range claim.Messages() {key := string(message.Key)val := string(message.Value)log4go.Info("%s-%s", key, val)session.MarkMessage(message, "")}return nil
}