1.首先初始化conf配置把kafka和ES的地址配置好还有一个日志方便查看
配置信息如下 用到的库是
2.读取conf配置存取进结构体
3.读取conf配置代码如下
4.完成了initConfig的初始化
5.初始化initLogger
6.初始化kafka
7.初始化ES
gopkg.in/olivere/elastic.v2 // 这个是操作ES的库
8.干活把kafka的数据写入ES
github.com/Shopify/sarama 这个是操作kafka的驱动库
上面代码是读kafka消费数据通过sendToES这个函数发送至ES里面
sendToES代码如下
Index就是索引名称
index().type().bodyjson().do()这样的写法是链式执行操作
9.写完了基本操作后 再写一个模拟写入数据进kafka的数据 代码如下
这个就是生产者往kafka里面写入数据进去消费10.我们启动我们的kafka 注意kafka依赖于zookeeper 先启动ZK然后启动kafka
我这里用的是zookeeper-3.4.12网上有下载
启动ZK
ZK已经成功启动
11.启动kafka 我这里是kafka_2.11-1.1.0
.\bin\windows\kafka-server-start.bat .\config\server.properties
kafka已经跑起来了
12.把kafka消费测试端也启动
.\bin\windows\kafka-console-consumer.bat --topic nginx_log --zookeeper 127.0.0.1 2181
消费端启动成功 一直等待数据进来消费
13.然后我们把ES 和Kib 都启动了
这是我们的ES版本是5.5.1的 已经跑起来了 接着启动我们的kib
kib里面有个配置config下面的叫kibana.yml里面设置好ES的地址和端口就处于监听ES状态
启动kib有点慢 稍微等一下就好
此时启动好了kib
14.测试kib是否启动
默认地址是http://localhost:5601
进入成功 确定没问题
15.编译我们的代码 写数据进kafka
从上面看我们知道一个再写 一个再消费
16.编译运行我们把kafka写入进ES里面的代码
运行了 这里就把kafka消费的数据 写入进ES里面
17.我们看一下数据是否有 进入kib
我们看到有数据了 已经成功了
好了 欢迎给赞!