1 简介

1.1 简介

Kafka is a distributed,partitioned,replicated commit logservice。它提供了类似于 JMS 的特性，但是在设计实现上完全不同，此外它并不是 JMS 规范的实现。kafka 对消息保存时根据 Topic 进行归类，发送消息者成为 Producer, 消息接受者成为 Consumer, 此外 kafka 集群有多个 kafka 实例组成，每个实例 (server) 成为 broker。无论是 kafka 集群，还是 producer 和 consumer 都依赖于 zookeeper 来保证系统可用性集群保存一些 meta 信息。

官网：http://kafka.apache.org/
中文网：https://www.orchome.com/5

在其中我们知道如果要 kafka 正常运行，必须配置 zookeeper，否则无论是 kafka 集群还是客户端的生存者和消费者都无法正常的工作的，使用需要先安装 zookeeper 集群：

Zookeeper 下载地址：http://zookeeper.apache.org/r…

1.2 参考资料

http://kafka.apache.org/intro…
https://www.cnblogs.com/likeh…

2 集群规划

2.1 机器规划

192.168.90.225 : zookeeper + kafka
192.168.90.226 : zookeeper + kafka
192.168.90.227 : zookeeper + kafka

2.2 目录规划

mkdir /usr/local/tmp    #临时目录，用于上传压缩包
mkdir  /topinfo  # 安装目录

3 zookeeper 集群

3.1 部署环境

操作系统：CentOS7 +
三台服务器：192.168.90.225/226/227
安装包：zookeeper-3.4.14.tar.gz

前置条件：jdk1.8+

3.2 安装 zookeeper

3.2.1 上传 zookeeper 安装包

在其中一台服务器上，上传安装包：

cd  /usr/local/tmp/   # 跳转到临时目录
rz   #通过 rz 上传 zookeeper-3.4.14.tar.gz

3.2.2 安装

1、解压

tar  -zxvf  zookeeper-3.4.14.tar.gz  # 解压
mv  zookeeper-3.4.14  zookeeper  #重命名
mv  zookeeper/  /topinfo/  # 将解压包拷贝到指定目录

2、进入 zookeeper 的目录

cd  /topinfo/zookeeper/  #进入目录
ll  # 列出目录的文件列表

如图：

3、修改配置文件
进入 conf 目录，复制 zoo_sample.cfg，并重命名为 zoo.cfg

cp  zoo_sample.cfg   zoo.cfg

4、打开并修改 zoo.cfg

vi  zoo.cfg

修改 dataDir 的存放目录，clientPort 默认保持不变，添加 zookeeper 的节点信息。

# dataDir 数据存放目录
dataDir= /topinfo/zookeeper/data
# the port at which the clients will connect
clientPort=2181
# zookeeper 集群的 ip:lead 和 folw 的通信端口: 选举端口
# server.index 为服务的 myid
server.0=192.168.90.225:2888:3888
server.1=192.168.90.226:2888:3888
server.2=192.168.90.227:2888:3888

wq! 退出保存后，创建数据的保持目录

mkdir -p /topinfo/zookeeper/data

在目录中创建 myid 文件（内容为 0）

touch /topinfo/zookeeper/data/myid
echo 0 > /topinfo/zookeeper/data/myid 

5、把 zookeeper 文件夹复制到另外两个服务器上，并创建相同的 dataDir 目录和 myid 文件。

scp -r /topinfo/zookeeper  root@192.168.90.226:/topinfo/

scp -r /topinfo/zookeeper  root@192.168.90.227:/topinfo/

注意：192.168.90.226 的 myid 文件内容为 1，192.168.90.227 的 myid 文件内容为 2，需要与 zoo.cfg 中的保持一致。

# 192.168.90.226 执行
mkdir -p /topinfo/zookeeper/data
touch /topinfo/zookeeper/data/myid
echo 1 > /topinfo/zookeeper/data/myid  

# 192.168.90.227 执行
mkdir -p /topinfo/zookeeper/data
touch /topinfo/zookeeper/data/myid
echo 2 > /topinfo/zookeeper/data/myid  

6、关闭防火墙

# 关闭防火墙
service iptables stop 
# 启动防火墙
service iptables start 
# 重启防火墙
service iptables restart 
# 查看防火墙状态
service iptables status

3.2.3 设置环境变量

设置环境变量

# 打开 profile
vim /etc/profile

# 加入以下内容
export ZK_HOME=/topinfo/zookeeper
export PATH=$ZK_HOME/bin:$PATH

# 保持退出后，执行以下命令，立即生效
source /etc/profile

3.2.4 启动

进入 bin 目录，分别在三台机器上启动 zookeeper：

zkServer.sh  start

启动成功后：
ZooKeeper JMX enabled by default
Using config: /topinfo/zookeeper/bin/../conf/zoo.cfg
Starting zookeeper … STARTED

在三台服务器上执行命令，查看启动状态：

zkServer.sh status

可以查看服务器的角色：leader / follower
如：
ZooKeeper JMX enabled by default
Using config: /topinfo/zookeeper/bin/../conf/zoo.cfg
Mode: follower

3.2.5 验证

查看 zookeeper 的状态

4 kafka 集群

4.1 部署环境

操作系统：CentOS7 +
三台服务器：192.168.90.225/226/227
安装包：kafka_2.11-2.3.0.tgz

4.2 安装 kafka

4.2.1 上传 kafka 安装包

在其中一台服务器上，上传安装包：

cd  /usr/local/tmp/   # 跳转到临时目录
rz   #通过 rz 上传

4.2.2 安装
1、解压

tar  -zxvf  kafka_2.11-2.3.0.tgz  # 解压
mv  kafka_2.11-2.3.0  kafka  #重命名
mv  kafka/  /topinfo/  # 将解压包拷贝到指定目录

2、修改 server.properties
进入 kafka 的 config 目录

cd  /topinfo/kafka/config

vi  server.properties  #修改配置

修改内容如下：

# broker 的 id
broker.id=0
# 修改监听的地址，ip 为本机的 ip
listeners=PLAINTEXT://192.168.90.225:9092
# 修改日志目录
log.dirs=/topinfo/kafka/data/kafka-logs
# 连接 zookeeper 集群
zookeeper.connect=192.168.90.225:2181,192.168.90.226:2181,192.168.90.227:2181
# 可删除 topic
delete.topic.enable=true

3、创建对应的日志目录

mkdir  -p  /topinfo/kafka/data/kafka-logs

4、将 kafka 文件夹复制到另外的两个节点上

scp  -r  /topinfo/kafka/ root@192.168.90.225:/topinfo/  # 远程复制

scp  -r  /topinfo/kafka/ root@192.168.90.226:/topinfo/  # 远程复制

mkdir  -p  /topinfo/kafka/data/kafka-logs  # 另外两台上创建目录

5、创建相同的 logDir 目录，且修改每个节点对应的 server.properties 文件的 broker.id:
192.168.90.225 为 1
192.168.90.227 为 2

vi  /topinfo/kafka/config/server.properties  #打开文件修改对应的 broker.id

4.2.3 设置环境变量

设置环境变量

# 打开 profile
vim /etc/profile

# 加入以下内容
export KAFKA_HOME=/topinfo/kafka
export PATH=$KAFKA_HOME/bin:$PATH

# 保持退出后，执行以下命令，立即生效
source /etc/profile

4.2.4 启动 kafka

注：启动时：先启动 zookeeper，后启动 kafka；关闭时：先关闭 kafka，后关闭 zookeeper

1、分别在每个节点上执行命令，启动 zookeeper，如果已经启动或，则跳过

zkServer.sh  start

2、启动 kafka

kafka-server-start.sh  /topinfo/kafka/config/server.properties &

4.3 测试

4.3.1 创建 topic

kafka-topics.sh --create --zookeeper 192.168.90.225:2181,192.168.90.226:2181,192.168.90.227:2181 --replication-factor 3 --partitions 3 --topic test

4.3.2 显示 topic 信息

kafka-topics.sh --describe --zookeeper 192.168.90.225:2181,192.168.90.226:2181,192.168.90.227:2181 --topic test
 

4.3.3 列出 topic

kafka-topics.sh --list --zookeeper 192.168.90.225:2181,192.168.90.226:2181,192.168.90.227:2181

4.3.4 创建生产者

在 master 节点上 测试生产消息

kafka-console-producer.sh  --broker-list  192.168.90.225:9092  -topic  test

4.3.5 创建消费者

在 worker 节点上 测试消费

kafka-console-consumer.sh --bootstrap-server 192.168.90.225:9092 -topic test --from-beginning

4.3.6 删除 topic 和关闭服务

删除 topic

kafka-topics.sh --delete --zookeeper 192.168.90.225:2181,192.168.90.226:2181,192.168.90.227:2181 --topic test

关闭 kafka 服务，在三台机器上执行 kafka-server-stop.sh 命令：

kafka-server-stop.sh

关闭 zookeeper:

zkServer.sh  stop