折腾 kafka

Apache Kafka

Apache Kafka 是一个分布式发布 - 订阅消息系统和一个强大的队列，可以处理大量的数据，并使您能够将消息从一个端点传递到另一个端点。

• 特点
  • 分布系统
  • 订阅消息
  • 强大队列
  • 数据落地到硬盘

安装

• 前置条件：需要 zookeeper。
• 安装 zookeeper
  • 官方下载
  • 接下即安装（ java 生态就是好）
  • conf 下 sample 文件改成 zoo.cfg (默认单机模式)
  • zookeeper 常用命令

   //开启 zookeeper bin/zkServer.sh start //使用客户端连接 bin/zkCli.sh //远程连接 bin/zkCli.sh -server ip:port //停止 zookeeper bin/zkServer.sh stop 

  • 其他相关 zookeeper 不细讲用到在学习
• 安装 kafaka
  • 官方下载
  • 接下即安装（ java 生态就是好）
  • kafaka 常用命令

   //开启 kafaka bin/kafka-server-start.sh config/server.properties //创建 Topic bin/kafka-topics.sh --create --zookeeper 192.168.112.20:2181 --replication-factor 1 --partitions 1 --topic test //Topic 展示 bin/kafka-topics.sh --list --zookeeper 192.168.112.20:2181 

遇到的问题：下载的是 kafka 的源码包，执行报错 classpath is empty. please build the project first e.g. by running 'gradlew jarall' 应该下载二进制包 下载地址：https://kafka.apache.org/downloads

使用 Go 实现生产消费

• 实现生产者

package main import ( "bufio" "fmt" "github.com/Shopify/sarama" "os" "strings" ) func main() { config := sarama.NewConfig() config.Producer.Return.Successes = true config.Producer.RequiredAcks = sarama.WaitForAll config.Producer.PartitiOner= sarama.NewRandomPartitioner //初始化生产者 producer, err := sarama.NewSyncProducer([]string{"192.168.112.20:9092"}, config) if err != nil { panic(err) } defer producer.Close() msg := &sarama.ProducerMessage{ Topic: "testGo", Partition: int32(-1), Key: sarama.StringEncoder("key"), } var value string for { // 生产消息 inputReader := bufio.NewReader(os.Stdin) value, err = inputReader.ReadString('\n') if err != nil { panic(err) } value = strings.Replace(value, "\n", "", -1) msg.Value = sarama.ByteEncoder(value) //发送消息 paritition, offset, err := producer.SendMessage(msg) if err != nil { fmt.Println("Send Message Fail") } //输出结果 fmt.Printf("Partion = %d, offset = %d\n", paritition, offset) } } 

• 实现消费者

package main import ( "fmt" "github.com/Shopify/sarama" "sync" ) //协成锁 等待所有协成执行完毕退出 var wg sync.WaitGroup func main() { //消费者初始化 consumer, err := sarama.NewConsumer([]string{"192.168.112.20:9092"}, nil) if err != nil { panic(err) } partitionList, err := consumer.Partitions("testGo") if err != nil { panic(err) } for partition := range partitionList { pc, err := consumer.ConsumePartition("testGo", int32(partition), sarama.OffsetNewest) if err != nil { panic(err) } 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() consumer.Close() } } 

遇到问题 github.com/Shopify/sarama 包在 windows 环境下需要 gcc，懒得搞直接虚拟机。

总结

利用 go + kafka 实现生产者消费模型非常简单和快速，在日常业务上使用消息队列可以考虑 kafka+go 快速迭代，后续应用上采坑在来分享。