搭建kafka集群
Kafka官方文档：http://kafka.apache.org
1. 安装java
Elasticsearch至少需要Java 8.版本的java，建议使用Oracle JDK 1.8.0_131版（官网建议）。如果没符合标准，请自行升级。本文只描述新安装java（由于本次安装kafka和elasticsearch在一台服务器上所以就直接安装java8）
[root@es-1 ~]# yum install -y java-1.8.0
[root@es-1 ~]# java -version
openjdk version "1.8.0_171"
OpenJDK Runtime Environment (build 1.8.0_171-b10)
OpenJDK 64-Bit Server VM (build 25.171-b10, mixed mode)
2. 下载安装kafka****（kafka_2.11-1.1.0****）
官方地址：https://www.apache.org/dyn/closer.cgi?path=/kafka/1.1.0/kafka_2.11-1.1.0.tgz
百度云：链接：https://pan.baidu.com/s/1yW3h9P9sMvkBJP46mHIr7w 密码：57e1
3. 上传、解压
上传到/opt/elk目录
[root@es-1 ~]# cd /opt/elk/
[root@es-1 elk]# ls
kafka_2.11-1.1.0.tgz
[root@es-1 elk]# tar -xvf kafka_2.11-1.1.0.tgz
4.****相关介绍
kafka 采用集群方式部署，测试环境部署，所以我用两台机器模拟出3个节点的cluster 。kafka 依赖zookeeper，zookeeper 简称ZK，已经包含在kafka的tar包里，不需要而外下载安装，为啥要用三个节点呢？这个是zk的一个选举leader的特性，官方推荐是奇数个server，奇数个最少3个，这样坏一个还有两个，可以正常选举。（官方解释：对于复制模式，需要最少三个服务器，并且强烈建议您有奇数个服务器。如果只有两个服务器，那么，如果有一个服务器失败，则没有足够的机器形成多数法定人数。两个服务器本质上不如单个服务器稳定，因为有两个单点故障。）
由于是集群配置，配置之间几乎差不多，所以配置好一个，直接scp到其余节点就ok了，到了其余节点把关键参数修改一下就行了。
5. 配置zookeeper
①给zookeeper 创建data 目录和logs 目录
[root@es-1 ~]# mkdir /opt/elk/kafka/zookeeper/{data,logs} -p
②创建myid 文件
[root@es-1 ~]# echo 1 > /opt/elk/kafka/zookeeper/data/myid #集群其他服务器需要改动
③配置zookeeper配置文件
[root@es-1 ~]# cd /opt/elk/kafka_2.11-1.1.0/config/
[root@es-1 config]# vim zookeeper.properties
#客户端连接端口
maxClientCnxns=0
#zk存放数据的目录，zk 需要有一个叫做myid的文件也是放到（必须）这个目录下
dataDir=/opt/elk/kafka/zookeeper/data
dataLogDir=/opt/elk/kafka/zookeeper/logs
clientPort=2181
#最大客户端连接数
maxClientCnxns=20
#是作为Zookeeper服务器之间或客户端与服务器之间维持心跳的时间间隔
tickTime=2000
#此配置表示，允许follower(相对于Leaderer言的“客户端”)连接并同步到Leader的初始化连接时间，以tickTime为单位。当初始化连接时间超过该值，则表示连接失败。
initLimit=10
#此配置项表示Leader与Follower之间发送消息时，请求和应答时间长度。如果follower在设置时间内不能与leader通信，那么此follower将会被丢弃。
syncLimit=5
#server.myid=ip:followers_connect to the leader:leader_election # server 是固定的，myid 是需要手动分配，第一个端口是follower是链接到leader的端口，第二个是用来选举leader 用的port
server.1=192.168.205.155:2888:3888
server.2=192.168.205.156:2888:3888
server.3=192.168.205.157:2888:3888
注意：如果有防火墙记得开放相应端口
6. 配置kafka
[root@es-1 ~]# cd /opt/elk/kafka_2.11-1.1.0/config/
[root@es-1 config]# vim server.properties
#每个server需要单独配置broker id，如果不配置系统会自动配置。
broker.id=0 ##集群其他服务器需要改动
#消费者的访问端口，logstash或者elasticsearch
listeners=PLAINTEXT://192.168.205.155:9092 ##集群其他服务器需要改动
#接收和发送网络信息的线程数
num.network.threads=3
#服务器用于处理请求的线程数，其中可能包括磁盘I/O。
num.io.threads=8
#套接字服务器使用的发送缓冲区(SO_SNDBUF)
socket.send.buffer.bytes=102400
#套接字服务器使用的接收缓冲区(SO_RCVBUF)
socket.receive.buffer.bytes=102400
#套接字服务器将接受的请求的最大大小(防止OOM)。
socket.request.max.bytes=104857600
#以逗号分隔的目录列表，其中存储日志文件。
log.dirs=/usr/local/kafka/logs #此目录要先创建，不会自动创建。（如果不行就关闭此选项）
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
zookeeper.connect=192.168.205.155:2181,192.168.205.156:2181,192.168.205.157:2181
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=0
7. 创建kafka 日志文件
[root@es-1 ~]# mkdir -p /usr/local/kafka/logs/
8.****配置集群其他节点
只需要把配置好的安装包直接分发到不同的机器上，然后修改zookeeper的myid，kafka的broker.id和listeners就可以了。
9. 启动kafka 集群（顺序是先启动zk ，然后再启动kafka ）
①启动zk
[root@es-1 ~]# nohup /opt/elk/kafka_2.11-1.1.0/bin/zookeeper-server-start.sh zookeeper.properties >>/dev/null 2>&1 &
[root@es-2 ~]# nohup /opt/elk/kafka_2.11-1.1.0/bin/zookeeper-server-start.sh zookeeper.properties >>/dev/null 2>&1 &
[root@es-3 ~]# nohup /opt/elk/kafka_2.11-1.1.0/bin/zookeeper-server-start.sh zookeeper.properties >>/dev/null 2>&1 &
②验证是否正确
#安装nc
[root@es-1 ~]# yum -y install nc
\#使用echo ruok|nc 127.0.0.1 2181 测试是否启动了该Server，若回复imok表示已经启动。
[root@es-1 ~]# echo ruok|nc 127.0.0.1 2181
imok[root@es-1 ~]#
#查看zk的配置，配置正常返回证明zk service 正常
[root@es-1 ~]# echo conf | nc 127.0.0.1 2181
clientPort=2181
dataDir=/opt/elk/kafka/zookeeper/data/version-2
dataLogDir=/opt/elk/kafka/zookeeper/logs/version-2
tickTime=2000
maxClientCnxns=20
minSessionTimeout=4000
maxSessionTimeout=40000
serverId=1
initLimit=10
syncLimit=5
electionAlg=3
electionPort=3888
quorumPort=2888
peerType=0
#stat 可以查看集群状态
[root@es-1 ~]# echo stat|nc 127.0.0.1 2181
Zookeeper version: 3.4.10-39d3a4f269333c922ed3db283be479f9deacaa0f, built on 03/23/2017 10:13 GMT
Clients:
/127.0.0.1:45542[0](queued=0,recved=1,sent=0)
Latency min/avg/max: 0/0/11
Received: 1025
Sent: 1035
Connections: 1
Outstanding: 0
Zxid: 0x100000074
Mode: follower
Node count: 28
#通过lsof 命令查看端口是否正常启动
[root@es-1 ~]# lsof -i:2181
COMMAND PID USER FD TYPE DEVICE SIZE/OFF NODE NAME
java 29022 root 94u IPv6 496186 0t0 TCP *:eforward (LISTEN)
# 还有如下常用命令：
ZooKeeper 支持某些特定的四字命令字母与其的交互。它们大多是查询命令，用来获取 ZooKeeper 服务的当前状态及相关信息。用户在客户端可以通过 telnet 或 nc 向 ZooKeeper 提交相应的命令
可以通过命令：
echo stat|nc 127.0.0.1 2181 来查看哪个节点被选择作为follower或者leader
echo ruok|nc 127.0.0.1 2181 测试是否启动了该Server，若回复imok表示已经启动。
echo dump| nc 127.0.0.1 2181 ,列出未经处理的会话和临时节点。
echo kill | nc 127.0.0.1 2181 ,关掉server
echo conf | nc 127.0.0.1 2181 ,输出相关服务配置的详细信息。
echo cons | nc 127.0.0.1 2181 ,列出所有连接到服务器的客户端的完全的连接 / 会话的详细信息。
echo envi |nc 127.0.0.1 2181 ,输出关于服务环境的详细信息（区别于 conf 命令）。
echo reqs | nc 127.0.0.1 2181 ,列出未经处理的请求。
echo wchs | nc 127.0.0.1 2181 ,列出服务器 watch 的详细信息。
echo wchc | nc 127.0.0.1 2181 ,通过 session 列出服务器 watch 的详细信息，它的输出是一个与 watch 相关的会话的列表。
echo wchp | nc 127.0.0.1 2181 ,通过路径列出服务器 watch 的详细信息。它输出一个与 session 相
③启动kafka
[root@es-1 ~]#nohup /opt/elk/kafka_2.11-1.1.0/bin/kafka-server-start.sh /opt/elk/kafka_2.11-1.1.0/config/server.properties >>/dev/null 2>&1 &
[root@es-2 ~]# nohup /opt/elk/kafka_2.11-1.1.0/bin/kafka-server-start.sh /opt/elk/kafka_2.11-1.1.0/config/server.properties >>/dev/null 2>&1 &
[root@es-3 ~]# nohup /opt/elk/kafka_2.11-1.1.0/bin/kafka-server-start.sh /opt/elk/kafka_2.11-1.1.0/config/server.properties >>/dev/null 2>&1 &
④检测kafka
#测试kafka 工作是否正常，新建一个topic
[root@es-1 bin]# /opt/elk/kafka_2.11-1.1.0/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
Created topic "test".
提示以下内容证明没有问题 Created topic "test".
#通过list 来查看
[root@es-1 ~]# /opt/elk/kafka_2.11-1.1.0/bin/kafka-topics.sh --zookeeper 192.168.205.155:2181 --list
test
⑤报错解决
1）如果无法启动zk集群或者kafka集群
解决办法：
配置文件可能写错，注意myid和配置文件要一致。
或者看kafka配置文件的log.dirs=对应的目录是否存在
上一篇：filebeat+ELK+kafka集群搭建（二：filebeat部署）
下一篇：filebeat+ELK+kafka集群搭建（四：logstash部署）
