kafka

本小节主要介绍如何使用pan快速完成kafka消息的生产，在此不再对kafka做相关的介绍，不太了解kafka的可以去kafka官网阅读相关文档。

1、启动zookeeper

  1. ./bin/zookeeper-server-start /usr/local/etc/kafka/zookeeper.properties

2、启动kafka

  1. ./bin/kafka-server-start /usr/local/etc/kafka/server.properties

3、创建topic

  1. ./bin/kafka-topics  --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic test

4、修改pan中kafka相关配置

  1. [KafkaProxy]
  2. enable=true//开关
  3. KafkaWaitAll=true
  4. KafkaCompression=true
  5. KafkaPartitioner=round
  6. KafkaProducerTimeout=10
  7. brokers=localhost:9092
  8. sasl=false
  9. user=
  10. password=
  11. valid= //topic白名单，若为空则所有topic均可发送
  12. failMode=retry/save/discard(无限次重试、保存到redis、丢弃)

配置文档请点击这里

5、编译

  1. make

6、运行

  1. ./bin/pan -c ../conf/conf.ini

7、业务代码

业务方往pan中发消息，需要引入kafkautil，具体使用方法如下

使用方法

  1. package main
  3. import (
  4.     "fmt"
  5.     "time"
  7.     "github.com/tal-tech/xtools/kafkautil"
  9.     "github.com/spf13/cast"
  10. )
  12. func main() {
  13.     t := time.Tick(5 * time.Second)
  14.     count := 0
  15.     for {
  16.         select {
  17.         case <-t:
  18.             count++
  19.             err := kafkautil.Send2Proxy("test", []byte("kafka "+cast.ToString(count)))
  20.             if err != nil {
  21.                 fmt.Println(err)
  22.             }
  23.             continue
  24.         }
  25.     }
  26. }

使用配置

  1. [KafkaProxy]
  2. unix=/home/www/pan.xesv5.com/pan.sock   //pan的sock文件地址
  3. host=localhost:9999  //pan的ip:port地址

注意事项

注意go.mod文件中替换包

  1. replace github.com/henrylee2cn/teleport v5.0.0+incompatible => github.com/hhtlxhhxy/github.com_henrylee2cn_teleport v1.0.0
  3. 或
  5. replace github.com/henrylee2cn/teleport v0.0.0 => github.com/hhtlxhhxy/github.com_henrylee2cn_teleport v1.0.0