一、实验介绍

1.1实验内容

实验在Hadoop集群上部署Kafka分布式发布订阅消息系统，并完成kafka消息管理验证。

1.2实验知识点

Kafka集群部署
Kafka消息处理流程

1.3实验环境

Kafka2.11
网易云平台

1.4实验资源

资源名称	存储目录
Kafka安装包	/opt/software/package/

1.5实验步骤清单

kafka安装部署（安装kafka、配置kafka集群环境）
kafka消息管理（master主机创建topic、查看topic、创建生产者、创建消费者）

二、实训架构

序号	IP地址	机器名	类型
1	172.25.10.140	master	Producer
2	172.25.10.141	slave1	Consumer
3	172.25.10.142	slave2	Consumer
4	172.25.10.143	slave2	Consumer

三、实验环境准备

进入实验列表，找到Kafka，点击开启实验，启动master、slave1及salve2，slave3。

四、实验步骤

4.1安装部署

4.1.1配置hosts文件及zookeeper的myid

#hosts文件末尾加入如下内容(masterip、slave1ip及slave2ip是这三台机器的ip地址

#slave1修改为2
[root@slave1 ~]# echo 2 >/usr/local/zookeeper/data/myid
#slave2修改为3
[root@slave2 ~]# echo 3 > /usr/local/zookeeper/data/myid
#slave2修改为4
[root@slave3 ~]# echo 4 > /usr/local/zookeeper/data/myid

#让三台机器profile环境变量生效
source /etc/profile

4.1.2安装Kafka

#解压安装Kafka
tar zxvf /opt/software/package/kafka_2.11-2.1.1.tgz -C /usr/local/

#重命名kafka
mv /usr/local/kafka_2.11-2.1.1/ /usr/local/kafka

#配置master kafka
vim /usr/local/kafka/config/server.properties
找到下面两行代码并分别注释
#broker.id=0
#zookeeper.connect=localhost:2181

在文件底部添加如下三个配置

broker.id=1
zookeeper.connect=master:2181,slave1:2181,slave2:2181,slave3:2181
listeners = PLAINTEXT://master:9092

配置说明
broker.id：每台机器不能一样
zookeeper.connect：因为实验环境中有3台zookeeper服务器，所以在这里zookeeper.connect设置为3台，必须全部加进去
listeners：Socket服务器侦听的地址，在配置集群的时候必须设置。

4.1.3配置kafka集群环境

1、拷贝kafka到另外三台服务器
scp -r /usr/local/kafka root@slave1:/usr/local/
scp -r /usr/local/kafka root@slave2:/usr/local/
scp -r /usr/local/kafka root@slave3:/usr/local/

2、修改这三台服务器的broker.id和listeners
#修改slave1
vim /usr/local/kafka/config/server.properties

broker.id=2
zookeeper.connect=master:2181,slave1:2181,slave2:2181,slave3:2181
listeners = PLAINTEXT://slave1:9092

#修改slave2
vim /usr/local/kafka/config/server.properties

broker.id=3
zookeeper.connect=master:2181,slave1:2181,slave2:2181,slave3:2181
listeners = PLAINTEXT://slave2:9092

#修改slave3
vim /usr/local/kafka/config/server.properties

broker.id=4
zookeeper.connect=master:2181,slave1:2181,slave2:2181,slave3:2181
listeners = PLAINTEXT://slave3:9092

3、启动zookeeper
#启动四台虚拟主机的zookeeper
zkServer.sh start
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/bin/…/conf/zoo.cfg
Starting zookeeper … STARTED

4、启动kafka
#启动四台虚拟主机的kafka
启动之前先且到bin目录下，用如下命令：cd /usr/local/kafka/bin
./kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties

5、jps命令检查是否启动成功
#Master主机检查

#Slave1主机检查

#Slave2主机检查

#Slave3主机检查

4.2 Kafka消息管理

4.2.1 master主机创建topic

/usr/local/kafka/bin/kafka-topics.sh --create --zookeeper master:2181 --replication-factor 1 --partitions 1 --topic test
如果成功的话，会输出：Created topic “test”.

4.2.2查看topic

#在master主机查看
/usr/local/kafka/bin/kafka-topics.sh --list --zookeeper master:2181

#在Slave1主机查看
/usr/local/kafka/bin/kafka-topics.sh --list --zookeeper master:2181

4.2.3创建生产者

#在master创建发布者
/usr/local/kafka/bin/kafka-console-producer.sh --broker-list master:9092 --topic test
#生产消息

test hello
test world

4.2.4创建消费者

#在slave2上创建消费者
/usr/local/kafka/bin/kafka-console-consumer.sh --bootstrap-server master:9092 --topic test --from-beginning

五、实验总结

Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理，也是为了通过集群来提供实时的消息。Kafka的使用依赖于zookeeper,安装Kafka前必须先安装zookeeper。Kafka集群包含一个或多个服务器，这种服务器被称为broker,Producer负责发布消息到Kafka broker,Consumer消息消费者，从Kafka broker读取消息的客户端。