Filebeat部署+Kafka接收消息

一、下载与解压Filebeat

1.Filebeat官方下载地址：https://www.elastic.co/cn/downloads/past-releases#filebeat
我下的是7.12版本,链接: link

2.上传压缩包

解压并重命名文件夹
tar -zxvf filebeat-7.12.0-linux-x86_64.tar.gz
mv filebeat-7.12.0-linux-x86_64 filebeat

home是用户目录，个人习惯放在这个，你们也可以放在别的目录里。

二、配置Filebeat

修改filebeat配置文件，把filebeat收集到的日志保存到kafka消息队列中。
只需要修改input和output配置。
input配置（输入配置）：
filebeat的输入配置在filebeat.inputs节点下。type为log，表明这是从文件获取输入。我们需要将log类型下的enabled的值置为true，在paths中编辑需要获取输入的文件，也可以使用通配符。例如采集的是日志文件。
进入filebeat目录编辑filebeat.yml文件

采集的日志是自己创建的，后续添加信息可以被kafka实时接收到。

将path原来的路径注释掉。
output配置（输出配置）：
我们需要将读取到的日志输出到Kafka集群中，而配制文件中默认没有Kafka集群的输出配置，因此需要在其中增加，并将output.elasticsearch和output.logstash都给注释掉。我们增加的Kafka集群配置如下图所示：

kafka中如果不存在这个topic，则会自动创建。

启动filebeat

./filebeat -e -c filebeat.yml

启动Kafka

注意：在启动kafka前先启动ZooKeeper，再开启一个session会话

1. ZooKeeper启动:
   进入ZooKeeper的bin目录
   ./zkServer.sh start
   
   使用netstat -ano | grep 2181 命令监听端口
   
2. kafka启动:
   进入 bin目录:
   ./kafka-server-start.sh …/config/server.properties 启动服务
   
3. 查看topic列表

./kafka-topics.sh --list --zookeeper 192.168.192.132:2181

可以看到产生了新的topic为newtest，接下来启动一个消费者去查看filebeat发送过来的消息，能看到消息说明我们的filebeat的output.kafka配置成功。
./kafka-console-consumer.sh --bootstrap-server 192.168.192.132:9092 --topic test --from-beginning

配置完成