ELK安装、部署、调试（四）KAFKA消息队列的安装和部署

1.简介

Kafka是一种高吞吐量的分布式发布订阅消息系统，它可以处理消费者在网站中的所有动作流数据。 这种动作（网页浏览，搜索和其他用户的行动）是在现代网络上的许多社会功能的一个关键因素。 这些数据通常是由于吞吐量的要求而通过处理日志和日志聚合来解决。

Kafka [1]  是一种高吞吐量 [2]  的分布式发布订阅消息系统，有如下特性：

通过O(1)的磁盘数据结构提供消息的持久化，这种结构对于即使数以TB的消息存储也能够保持长时间的稳定性能。

高吞吐量 [2]  ：即使是非常普通的硬件Kafka也可以支持每秒数百万 [2]  的消息。

1.1 使用消息队列好处
1)、提高扩展性：因为消息队列解耦了处理过程，有新增需求时只要另外增加处理过程即可。

2)、提高峰值处理能力：在访问量剧增的情况下，应用仍然需要继续发挥作用。使用消息队列能够使关键组件顶住突发的访问压力，而不会因为突发的超负荷请求而完全崩溃；

3）、提高系统的可恢复性：消息队列降低了进程间的耦合度，所以即使一个处理消息的进程挂掉，加入队列中的消息仍然可以在系统回复后被处理
1.2 kafka的结构

• Producer：消息生产者，产生的消息将会被发送到某个topic
• Consumer：消息消费者，消费的消息内容来自某个topic
• Topic：消息根据topic进行归类，topic其本质是一个目录，即将同一主题消息归类到同一个目录
• Broker：每一个kafka实例（或者说每台kafka服务器节点）就是一个broker，一个broker可以有多个topic

2.下载

获取我的baidu网盘中ELK
kafka.apache.org 官方网址
由于我下载的filebeat是7.9.3，与这个版本filebeat配合的kafka版本为0.11--2.2.2
https://www.elastic.co/guide/en/beats/filebeat/7.9/kafka-output.html 这里有说明。

Compatibility
This output works with all Kafka versions in between 0.11 and 2.2.2. Older versions might work as well, but are
not supported.

3.安装

tar zxvf kafka_2.11-2.0.1.tgz -C /usr/local
ls
cd /usr/local/
mv kafka_2.11-2.0.1 kafka
./bin 执行文件
./config 配置文件主配置文件为 server.properties

4.配置

vi server.properties

broker.id=0 集群内的唯一标识
listeners=PLAINTEXT://ip[或者主机名]:9092          #配置监听端口  9092位kafka默认端口
log.dirs=/usr/local/kafka/logs      #日志及传输的数据存放目录可以通过“，”分隔存放到多个目录，如
log.dirs=/data1,/data2
num.partitions=6  #定义新的topic有多少个分区
num.retention.hours=60  消息日志保留的时间，单位是小时
#num.retention.minutes 或者num.retention.ms，可以多个参数同时设置，时间最小生效
log.segment.bytes=1073741824  每个段文件的大小，1G为默认值，topic--partition--segement
zookeeper.connect=10.10.10.71:2181,10.10.10.72:2181,10.10.10.73:2181
auto.create.topics.enable=true  #可以自动创建topics
delete.topic.enable=true  #可以自动删除空闲的topics，是逻辑的删除，非物理删除。
zookeeper.connect.timeout.ms=6000

mkdir /data1
mkdir /data2

5.启动kafka 

cd /usr/local/kafka
./bin/kafka-server-start.sh ./config/server.properties -d      #-d 后台执行
或者nohup ./kafka-server-start.sh ../config/server.properties &

启动时可能会报错，注意关闭服务器的防火墙

jps查看

[root@localhost kafka]# jps
23696 Kafka
9271 QuorumPeerMain
24077 Jps

kafka提供了多个命令来查看，创建，修改，删除topic信息，也可以通过命令来测试日志消息
命令存放路径为usr/local/kafka/bin

[root@localhost bin]# pwd
/usr/local/kafka/bin

[root@localhost bin]# ls
connect-distributed.sh        kafka-consumer-groups.sh     kafka-preferred-replica-election.sh  kafka-streams-
application-reset.sh  zookeeper-server-start.sh
connect-standalone.sh         kafka-consumer-perf-test.sh  kafka-producer-perf-test.sh          kafka-topics.sh  
                   zookeeper-server-stop.sh
kafka-acls.sh                 kafka-delegation-tokens.sh   kafka-reassign-partitions.sh         kafka-
verifiable-consumer.sh        zookeeper-shell.sh
kafka-broker-api-versions.sh  kafka-delete-records.sh      kafka-replica-verification.sh        kafka-
verifiable-producer.sh
kafka-configs.sh              kafka-dump-log.sh            kafka-run-class.sh                   trogdor.sh
kafka-console-consumer.sh     kafka-log-dirs.sh            kafka-server-start.sh                windows
kafka-console-producer.sh     kafka-mirror-maker.sh        kafka-server-stop.sh                 zookeeper-
security-migration.sh

例子：kafka的任何操作都需要zookeeper
查看有多少个topic

./kafka-topics.sh  --zookeeper 10.10.10.71:2181,10.10.10.72:2181,10.10.10.73:2181 --list

创建topic

./kafka-topics.sh  --create --zookeeper 10.10.10.71:2181,10.10.10.72:2181,10.10.10.73:2181 --replication-factor
1 --partitions 3 --topic mytopic

实例：

[root@localhost bin]# ./kafka-topics.sh  --create --zookeeper 10.10.10.71:2181,10.10.10.72:2181,10.10.10.73:2181
--replication-factor 1 --partitions 3 --topic mytopic
Created topic "mytopic".
[root@localhost bin]# ./kafka-topics.sh  --zookeeper 10.10.10.71:2181,10.10.10.72:2181,10.10.10.73:2181 --list  
           mytopic
[root@localhost bin]#

看看topic的具体信息

./kafka-topics.sh  --describe --zookeeper 10.10.10.71:2181,10.10.10.72:2181,10.10.10.73:2181  --topic mytopic
 
 
[root@localhost bin]# ./kafka-topics.sh  --describe --zookeeper
10.10.10.71:2181,10.10.10.72:2181,10.10.10.73:2181  --topic mytopic
Topic:mytopic   PartitionCount:3        ReplicationFactor:1     Configs:
        Topic: mytopic  Partition: 0    Leader: 1       Replicas: 1     Isr: 1
        Topic: mytopic  Partition: 1    Leader: 2       Replicas: 2     Isr: 2
        Topic: mytopic  Partition: 2    Leader: 3       Replicas: 3     Isr: 3

消费信息
交互式的生成消息到topic mytopic中

./kafka-console-producer.sh  --broker-list 10.10.10.71:9092,10.10.10.72:9092,10.10.10.73:9092 --topic mytopic

ctrl +c  结束查看消费消息

./kafka-console-consumer.sh  --bootstrap-server 10.10.10.71:9092,10.10.10.72:9092,10.10.10.73:9092 --topic
mytopic
 
./kafka-console-consumer.sh  --bootstrap-server 10.10.10.71:9092,10.10.10.72:9092,10.10.10.73:9092 --topic
mytopic
 --from-beginning  #从头开始消费
 

删除topic

[root@localhost bin]# ./kafka-topics.sh  --delete --zookeeper 10.10.10.71:2181,10.10.10.72:2181,10.10.10.73:2181
--topic mytopic
 
[root@localhost bin]# ./kafka-topics.sh  --delete --zookeeper 10.10.10.71:2181,10.10.10.72:2181,10.10.10.73:2181
--topic mytopic
Topic mytopic is marked for deletion.
Note: This will have no impact if delete.topic.enable is not set to true.