Kafka 集群部署

一、基础环境

    Servers：       192.168.255.128/129/130
    JAVA：           rpm -qa | grep java   >>   java-1.8.0-openjdk-1.8.0.161-0.b14.el7_4.x86_64
    Zookeeper：  http://mirrors.hust.edu.cn/apache/zookeeper/zookeeper-3.4.11/zookeeper-3.4.11.tar.gz
    Kafka：          http://mirrors.shu.edu.cn/apache/kafka/1.1.0/kafka_2.12-1.1.0.tgz

二、环境部署

注：各节点均需部署
2.1 安装 JAVA、Zookeeper 及 Kafka

[root@host128 ~]# yum -y install java
[root@host128 ~]# wget http://mirrors.hust.edu.cn/apache/zookeeper/zookeeper-3.4.11/zookeeper-3.4.11.tar.gz
[root@host128 ~]# wget http://mirrors.shu.edu.cn/apache/kafka/1.1.0/kafka_2.12-1.1.0.tgz
[root@host128 ~]# tar -zxf kafka_2.12-1.1.0.tgz && mv kafka_2.12-1.1.0 /opt/kafka
[root@host128 ~]# tar -zxf zookeeper-3.4.11.tar.gz && mv zookeeper-3.4.11 /opt/zookeeper

2.2 修改 Zookeeper 配置

#添加集群配置（server.0/1/2):
[root@host128 ~]# cat <<EOF>/opt/zookeeper/conf/zoo.cfg
tickTime=2000
initLimit=10
syncLimit=5
dataDir=/data/zookeeper
clientPort=2181
server.128=192.168.255.128:2888:3888
server.129=192.168.255.129:2888:3888
server.130=192.168.255.130:2888:3888
EOF

2.3 Kafka 配置

#添加解析
[root@host128 ~]# cat << EOF >>/etc/hosts
192.168.255.128 host128
192.168.255.129 host129
192.168.255.130 host130
EOF

2.4 防火墙配置

#zk:
[root@host128 ~]# firewall-cmd --permanent --add-port=2181/tcp #对cline端提供服务
[root@host128 ~]# firewall-cmd --permanent --add-port=2888/tcp #集群内机器通讯使用（Leader监听此端口）
[root@host128 ~]# firewall-cmd --permanent --add-port=3888/tcp #选举leader使用
#kafka:
[root@host128 ~]# firewall-cmd --permanent --add-port=9092/tcp #Kafka监听端口
#配置加载
[root@host128 ~]# firewall-cmd --reload
注：
#针对特定IP放开访问
[root@host128 ~]# firewall-cmd --permanent --add-rich-rule 'rule family=ipv4 source address=10.65.1.0/24 port port=2181 protocol=tcp accept'
[root@host128 ~]# firewall-cmd --permanent --add-rich-rule 'rule family=ipv4 source address=10.65.1.0/24 port port=2888 protocol=tcp accept'
[root@host128 ~]# firewall-cmd --permanent --add-rich-rule 'rule family=ipv4 source address=10.65.1.0/24 port port=3888 protocol=tcp accept'
[root@host128 ~]# firewall-cmd --permanent --add-rich-rule 'rule family=ipv4 source address=10.65.1.0/24 port port=9092 protocol=tcp accept'

注：各节点配置不同部分

2.5 Zk 配置及启动测试

#创建zookeeper的myid文件，取IP地址D位作为ID
[root@host128 ~]# export ID="$(ip a|grep "inet "|grep -v " lo"|awk '{print$2}'|awk -F"/" '{print$1}'|awk -F"." '{print$NF}')"
[root@host128 ~]# mkdir /data/zookeeper && echo "${ID}" > /data/zookeeper/myid
#各节点依次执行
[root@host128 ~]# zkServer.sh start

2.6 Kafka 配置及启动测试

#修改broker.id为ID地址，不可大于reserved.broker.max.id（1000）
[root@host128 ~]# sed -i "/broker.id/ s/0/${ID}/g" /opt/kafka/config/server.properties
各节点依次执行
[root@host128 ~]# /opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/server.properties &
...
...
[2018-04-17 21:43:32,246] INFO [/config/changes-event-process-thread]: Starting (kafka.common.ZkNodeChangeNotificationListener$ChangeEventProcessThread)
[2018-04-17 21:43:32,302] INFO Kafka version : 1.1.0 (org.apache.kafka.common.utils.AppInfoParser)
[2018-04-17 21:43:32,302] INFO Kafka commitId : fdcf75ea326b8e07 (org.apache.kafka.common.utils.AppInfoParser)
[2018-04-17 21:43:32,315] INFO [KafkaServer id=0] started (kafka.server.KafkaServer)
## 启动成功

2.7 systemd 配置

#Zookeeper
[root@host128 ~]# useradd zookeeper -U -d /opt/zookeeper/ -s /sbin/nologin
[root@host128 ~]# chown -R  zookeeper.zookeeper /opt/zookeeper/
[root@host128 ~]# chown -R  zookeeper.zookeeper /data/zookeeper/
[root@host128 ~]# cat << "EOF">/lib/systemd/system/zookeeper.service
[Unit]
Description=Zookeeper service
After=network.target
[Service]
Type=forking
Environment=ZHOME=/opt/zookeeper
Environment=ZOO_LOG_DIR=/opt/zookeeper
ExecStart=/opt/zookeeper/bin/zkServer.sh start \
  ${ZHOME}/conf/zoo.cfg
ExecStop=/opt/zookeeper/bin/zkServer.sh stop
Restart=always
User=zookeeper
Group=zookeeper
[Install]
WantedBy=multi-user.target
EOF
#Kafka
[root@host128 ~]# useradd kafka -U -d /opt/kafka/ -s /sbin/nologin
[root@host128 ~]# chown -R kafka.kafka /opt/kafka
[root@host128 ~]# mkdir /data/kafka-logs && chown -R kafka.kafka /data/kafka-logs/
[root@host128 ~]# cat <<"EOF">/lib/systemd/system/kafka.service
[Unit]
Description=Kafka service
After=network.target
[Service]
Type=forking
Environment=KHOME=/opt/kafka
ExecStart=/opt/kafka/bin/kafka-server-start.sh -daemon \
 ${KHOME}/config/server.properties
ExecStop=ExecStop=/opt/kafka/bin/kafka-server-stop.sh
Restart=always
User=kafka
Group=kafka
[Install]
WantedBy=multi-user.target
EOF
#启动服务及开机启动配置
[root@host128 ~]# systemctl start zookeeper && systemctl enable zookeeper
[root@host128 ~]# systemctl start kafka && systemctl enable kafka

三、测试

#生产者启动：
[root@host128 ~]# /opt/kafka/bin/kafka-console-producer.sh --broker-list 192.168.255.128:9092,192.168.255.129:9092,192.168.255.2 --topic test
>>test    ## 测试消息输入
>>
#消费者启动：
[root@host129 ~]# /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server 192.168.255.128:9092,192.168.255.129:9092,192.168.255.130:9092 --topic test
test      ## 消息输出