搭建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=对应的目录是否存在