二、Kafka集群部署

 

二 Kafka 集群部署

 

2.1 环境准备

2.1.1 集群规划

node01　　 node02 　　node03

zk 　　　　    zk 　　　　zk

kafka 　　   kafka 　　　kafka

2.1.2 jar 包下载 http://kafka.apache.org/downloads.html

2.1.3 虚拟机准备

1）准备 3 台虚拟机

2）配置 ip 地址———————————————————————————

3）配置主机名称

4）3 台主机分别关闭防火墙

[hadoop@node01 hadoop]# chkconfig iptables off

[hadoop@node02 hadoop]# chkconfig iptables off

[hadoop@node03 hadoop]# chkconfig iptables off

2.1.4 安装 jdk

2.1.5 安装 Zookeeper

　　0）集群规划 在 node01、node02 和 node03 三个节点上部署 Zookeeper。

　　1）解压安装

　　　　（1）解压 zookeeper 安装包到/home/hadoop/目录下 [hadoop@hadoop01 ~]$ tar -zxvf zookeeper-3.4.10.tar.gz -C /bd/

　　　　（2）在/bd/zk/这个目录下创建 zkData mkdir -p zkData

　　　　（3）重命名/bd/zk/conf 这个目录下的 zoo_sample.cfg 为 zoo.cfg        mv zoo_sample.cfg zoo.cfg

　　2）配置 zoo.cfg 文件

　　　　（1）具体配置 dataDir=/bd/zk/zkData

　　　　　　增加如下配置 #######################cluster##########################

　　　　　　　　server.2=node01:2888:3888

　　　　　　　　server.3=node02:2888:3888

　　　　　　　　server.4=node03:2888:3888

　　　　（2）配置参数解读 Server.A=B:C:D。

　　　　　　A 是一个数字，表示这个是第几号服务器；———————————————————————————

　　　　　　B 是这个服务器的 ip 地址；

　　　　　　 C 是这个服务器与集群中的 Leader 服务器交换信息的端口；

　　 　　　　D 是万一集群中的 Leader 服务器挂了，需要一个端口来重新进行选举，选出一个新的 Leader，而这个端口就是用来执行选举时服务器相互通信的端口

　　　　　　　　集群模式下配置一个文件 myid，这个文件在 dataDir 目录下，这个文件里面有一个数就是 A 的值，

　　　　　　　　Zookeeper 启动时读取此文件，拿到里面的数据与 zoo.cfg 里面的配置信息比 较从而判断到底是哪个 server。

　　3）集群操作

　　　　（1）在/bd/zk/zkData 目录下创建一个 myid 的文件

　　　　　　　　touch myid

　　　　　　　　添加 myid 文件，注意一定要在 linux 里面创建，在 notepad++里面很可能乱码

　　　　（2）编辑 myid 文件

　　　　　　　　vi myid

　　　　　　　　在文件中添加与 server 对应的编号：如 2

　　　　（3）拷贝配置好的 zookeeper 到其他机器上

　　　　　　　　scp -r zk/ hadoop@node02:/bd/

　　　　　　　　scp -r zk/ hadoop@node03:/bd/

　　　　　　　　并分别修改 myid 文件中内容为 3、4

　　　　（4）分别启动 zookeeper

　　　　　　　　[hadoop@node01 zk]# bin/zkServer.sh start

　　　　　　　　[hadoop@node02 zk]# bin/zkServer.sh start

　　　　　　　　[hadoop@node03 zk]# bin/zkServer.sh start

　　　　（5）查看状态

　　　　　　　　[hadoop@node01 zk]# bin/zkServer.sh status

　　　　　　　　JMX enabled by default

　　　　　　　　Using config: /bd/zk/bin/../conf/zoo.cfg

　　　　　　　　Mode: follower

　　　　　　　　[hadoop@node02 zk]# bin/zkServer.sh status

　　　　　　　　JMX enabled by default

———————————————————————————

　　　　　　Using config: /bd/zk/bin/../conf/zoo.cfg

　　　　　　Mode: leader

　　　　　　[hadoop@node03 zk]# bin/zkServer.sh status

　　　　　　JMX enabled by default

　　　　　　Using config: /bd/zk/bin/../conf/zoo.cfg

　　　　　　Mode: follower

2.2 Kafka 集群部署

　　1）解压安装包  　　

　　　　　　 [hadoop@hadoop01 ~]$ tar -zxvf kafka_2.11-0.11.0.0.tgz -C /bd/

　　2）修改解压后的文件名称

　　　　　　 [hadoop@hadoop01 bd]$ mv kafka_2.11-0.11.0.0/ kafka

　　3）在/bd/kafka 目录下创建 logs 文件夹

　　　　　　 [hadoop@hadoop01 kafka]$ mkdir logs

　　4）修改配置文件

　　　　　　 [hadoop@hadoop01 kafka]$ cd config/

　　　　　　[hadoop@hadoop01 config]$ vi server.properties

　　　　　　输入以下内容：

　　　　　　　　#broker 的全局唯一编号，不能重复

　　　　　　　　broker.id=0

　　　　　　　　#删除 topic 功能使能

　　　　　　　　delete.topic.enable=true

　　　　　　　　#处理网络请求的线程数量

　　　　　　　　num.network.threads=3

　　　　　　　　#用来处理磁盘 IO 的现成数

　　　　　　　　num.io.threads=8

　　　　　　　　#发送套接字的缓冲区大小

　　　　　　　　socket.send.buffer.bytes=10240

　　　　　　　　#接收套接字的缓冲区大小

———————————————————————————

　　　　　　　　socket.receive.buffer.bytes=102400

　　　　　　　　#请求套接字的缓冲区大小

　　　　　　　　 socket.request.max.bytes=104857600

　　　　　　　　#kafka 运行日志存放的路径

　　　　　　　　log.dirs=/bd/kafka/logs

　　　　　　　　#topic 在当前 broker 上的分区个数

　　　　　　　　 num.partitions=1

　　　　　　　　 #用来恢复和清理 data 下数据的线程数量

　　　　　　　　num.recovery.threads.per.data.dir=1

　　　　　　　　#segment 文件保留的最长时间，超时将被删除

　　　　　　　　 log.retention.hours=168

　　　　　　　　 #配置连接 Zookeeper 集群地址

　　　　　　　　zookeeper.connect=node01:2181,node02:2181,node03:2181

　　　 5）配置环境变量

　　　　　　　　[hadoop@node01 bd]# vi /etc/profile

　　　　　　　　　　#KAFKA_HOME

　　　　　　　　　　export KAFKA_HOME=/bd/kafka

　　　　　　　　　　export PATH=$PATH:$KAFKA_HOME/bin

　　　　　　　　[hadoop@node01 bd]# source /etc/profile

　　　6）分发安装包

　　　　　　　　[hadoop@node01 etc]# xsync profile

　　　　　　　　[hadoop@hadoop102 module]$ xsync kafka/

　　   7）分别在 node02 和 node03 上修改配置文件/bd/kafka/config/server.properties 中的 broker.id=1、 broker.id=2

　　　　　　　　 注：broker.id 不得重复

　　　8）启动集群

　　　　　　　　依次在 node01、node02、node03 节点上启动 kafka

　　　　　　　　[hadoop@node01 kafka]$ bin/kafka-server-start.sh config/server.properties &

　　　　　　　　[hadoop@node02kafka]$ bin/kafka-server-start.sh config/server.properties &

　　　　　　　　[hadoop@node03kafka]$ bin/kafka-server-start.sh config/server.properties &

　　　　　　注:kafka 后台运行的方法:

　　　　　　　　1. bin/kafka-server-start.sh config/server.properties &

　　　　　　　　2. bin/kafka-server-start.sh -daemon config/server.properties

　　　　　　　　3. bin/kafka-server-start.sh config/server.properties(先前台运行,ctrl+z,bg 命令后台运行，fg 命 令还可以切换回前台)

2.3 Kafka 命令行操作

　　1）查看当前服务器中的所有 topi

　　　　　　[hadoop@node01 kafka]$ bin/kafka-topics.sh --list --zookeeper node01:2181

　　2）创建 topic

　　　　　　[hadoop@node01 kafka]$bin/kafka-topics.sh --create --zookeeper node02:2181 --replication-factor 3 --partitions 1 --topic first

　　　　　　选项说明：

　　　　　　--topic 定义 topic 名

　　　　　　--replication-factor 定义副本数

　　　　　　--partitions 定义分区数

　　3） 删除 topic

　　　　　　 [hadoop@node01 kafka]$ bin/kafka-topics.sh --delete --zookeeper node01:2181 --topic first

　　　　　　需要 server.properties 中设置 delete.topic.enable=true 否则只是标记删除或者直接重启。

　　4）发送消息

　　　　　　[hadoop@node01 kafka]$ bin/kafka-console-producer.sh --broker-list node01:9092 --topic first >hello world >hadoop hadoop

　    5）消费消息

　　　　　　[hadoop@node01 kafka]$ bin/kafka-console-consumer.sh --zookeeper node01:2181 --from-beginning --topic first

　　 6）查看某个 Topic 的详情

　　　　　　[hadoop@node01 kafka]$ bin/kafka-topics.sh --topic first --describe --zookeeper node01:2181

2.4 Kafka 配置信息

2.4.1 Broker配置信息

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

2.4.2 Producer 配置信息

 

 

 

 

 

 

2.4.3 Consumer 配置信息