02-kafka集群搭建

0. 服务器准备

• 服务器规划

IP	hostname	服务	服务
10.10.xxx.61	kafka-01	zookeeper-01	kafka-01
10.10.xxx.62	kafka-01	zookeeper-02	kafka-02
10.10.xxx.63	kafka-01	zookeeper-03	kafka-03

• hostname
  三台服务器hostname设置分别如上表
• hosts
  /etc/hosts 文件添加如下内容

10.10.239.61     kafka-01
10.10.239.62     kafka-02
10.10.239.63     kafka-03

• java环境已安装

1. zookeeper集群

1.1 下载

下载地址： https://mirrors.bfsu.edu.cn/apache/zookeeper/

下载带bin的这个tar包，另一个现在的版本是源码。

下载错了报错如下

 [root@kafka-01 logs]# tailf zookeeper-root-server-kafka-01.out
错误: 找不到或无法加载主类 org.apache.zookeeper.server.quorum.QuorumPeerMain

1.2 拷贝

• 拷贝
  下载完毕，分别拷贝到3台服务器的/data目录（我服务器的数据盘），解压缩如下。

[root@kafka-01 data]# ls
apache-zookeeper-3.6.2-bin  apache-zookeeper-3.6.2-bin.tar.gz
[root@kafka-01 data]# pwd
/data

• 软连接
  因为目录放在数据盘且目录名保留了版本号，为了使用方便，我在/opt下创建一个软连接：

[root@kafka-01 data]# ln -s /data/apache-zookeeper-3.6.2-bin /opt/zookeeper
[root@kafka-01 opt]# ll
总用量 4
lrwxrwxrwx  1 root root   32 3月   2 13:28 zookeeper -> /data/apache-zookeeper-3.6.2-bin

1.3 配置变量

/etc/profile 文件添加如下内容：

###### zookeeper #############
export ZK_HOME=/opt/zookeeper
export PATH=$ZK_HOME/bin:$PATH

引用变量

[root@kafka-01 opt]# source /etc/profile

1.4 配置

• 复制配置文件

[root@kafka-01 opt]# cd /opt/zookeeper/conf/
[root@kafka-01 conf]# ll
总用量 16
-rw-r--r-- 1 cloud_user cloud_user  535 9月   4 20:43 configuration.xsl
-rw-r--r-- 1 cloud_user cloud_user 3435 9月   4 20:43 log4j.properties
-rw-r--r-- 1 cloud_user cloud_user 1148 9月   4 20:43 zoo_sample.cfg
[root@kafka-01 conf]# cp zoo_sample.cfg zoo.cfg

• 修改zoo.cfg 文件如下：

# 修改数据目录（这个目录稍后创建）
dataDir=/opt/zookeeper/data
#添加集群信息（后边myid需要和这个一致）。
server.1=10.10.xxx.61:2888:3888
server.2=10.10.xxx.62:2888:3888
server.3=10.10.xxx.63:2888:3888

• 创建数据目录

[root@kafka-01 conf]# cd /opt/zookeeper/
[root@kafka-01 zookeeper]# mkdir data

• myid
  给个节点写入myid，需要和前边配置文件中一致。三台分别写入1，2，3。

[root@kafka-01 zookeeper]# cd data/
[root@kafka-01 data]# echo 1 > myid

1.5 启动

• 启动

[root@kafka-01 data]# zkServer.sh start
[root@kafka-01 data]# netstat -ntlp
Active Internet connections (only servers)
Proto Recv-Q Send-Q Local Address           Foreign Address         State       PID/Program name
tcp        0      0 0.0.0.0:22              0.0.0.0:*               LISTEN      941/sshd
tcp        0      0 127.0.0.1:25            0.0.0.0:*               LISTEN      1392/master
tcp6       0      0 :::17822                :::*                    LISTEN      941/sshd
tcp6       0      0 :::37923                :::*                    LISTEN      17244/java
tcp6       0      0 :::2181                 :::*                    LISTEN      17244/java
tcp6       0      0 10.10.239.61:3888       :::*                    LISTEN      17244/java
tcp6       0      0 :::8080                 :::*                    LISTEN      17244/java
tcp6       0      0 :::22                   :::*                    LISTEN      941/sshd
tcp6       0      0 ::1:25                  :::*                    LISTEN      1392/master

• 设置开机启动

[root@kafka-01 init.d]# cat  > /etc/init.d/zookeeper.sh << EOF 
#!/bin/sh
#chkconfig: 2345 55 56
#description:zookeeper.sh
source /etc/profile
zkServer.sh start
EOF
[root@kafka-01 init.d]# chmod 755 /etc/init.d/zookeeper.sh
[root@kafka-01 init.d]# systemctl enable zookeeper.sh
zookeeper.sh.service is not a native service, redirecting to /sbin/chkconfig.
Executing /sbin/chkconfig zookeeper.sh on

说明：
脚本中引用一下profile文件，因为子shell中读不到JAVA_HOME 和 ZK_HOME

2. kafka

2.1 下载

• url
  http://kafka.apache.org/

2.2 拷贝到服务器

• 下载完毕后，拷贝到/data 目录下，并解压缩。结果如下：

[root@kafka-01 data]# ll
总用量 79224
drwxr-xr-x 8 root root     4096 3月   2 13:37 apache-zookeeper-3.6.2-bin
-rw-r--r-- 1 root root 12515974 3月   2 13:27 apache-zookeeper-3.6.2-bin.tar.gz
drwxr-xr-x 7 root root     4096 3月   2 15:28 kafka_2.13-2.7.0
-rw-r--r-- 1 root root 68583422 3月   2 15:07 kafka_2.13-2.7.0.tgz
drwx------ 2 root root    16384 3月   2 10:54 lost+found

• 为方便使用，在/opt下创建一个软连接

[root@kafka-01 data]# ln -s /data/kafka_2.13-2.7.0 /opt/kafka
[root@kafka-01 data]# cd /opt/
[root@kafka-01 opt]# ll
总用量 4
lrwxrwxrwx  1 root root   22 3月   2 15:33 kafka -> /data/kafka_2.13-2.7.0
lrwxrwxrwx  1 root root   32 3月   2 13:28 zookeeper -> /data/apache-zookeeper-3.6.2-bin

2.2 配置文件

• config目录

[root@kafka-01 config]# cd /opt/kafka/config/
[root@kafka-01 config]# ll
总用量 72
-rw-r--r-- 1 root root  906 12月 16 21:58 connect-console-sink.properties
-rw-r--r-- 1 root root  909 12月 16 21:58 connect-console-source.properties
-rw-r--r-- 1 root root 5321 12月 16 21:58 connect-distributed.properties
-rw-r--r-- 1 root root  883 12月 16 21:58 connect-file-sink.properties
-rw-r--r-- 1 root root  881 12月 16 21:58 connect-file-source.properties
-rw-r--r-- 1 root root 2247 12月 16 21:58 connect-log4j.properties
-rw-r--r-- 1 root root 2540 12月 16 21:58 connect-mirror-maker.properties
-rw-r--r-- 1 root root 2262 12月 16 21:58 connect-standalone.properties
-rw-r--r-- 1 root root 1221 12月 16 21:58 consumer.properties
-rw-r--r-- 1 root root 4674 12月 16 21:58 log4j.properties
-rw-r--r-- 1 root root 1925 12月 16 21:58 producer.properties
-rw-r--r-- 1 root root 6888 3月   2 15:20 server.properties
-rw-r--r-- 1 root root 1032 12月 16 21:58 tools-log4j.properties
-rw-r--r-- 1 root root 1169 12月 16 21:58 trogdor.conf
-rw-r--r-- 1 root root 1205 12月 16 21:58 zookeeper.properties

我们需要修改 server.properties文件，修改如下：

• 修改broker.id

三台不能一样，如分别是1，2，3

############################# Server Basics #############################
# The id of the broker. This must be set to a unique integer for each broker.
broker.id=1

• 链接zookeeper

zookeeper.connect=10.10.xxx.61:2181,10.10.xxx.62:2181,10.10.xxx.63:2181

• 修改后，过滤注释和空格查看内容如下：

[root@kafka-01 config]# cat server.properties |grep -v "^#"|grep -v "^$"
broker.id=1
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/tmp/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
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=10.10.239.61:2181,10.10.239.62:2181,10.10.239.63:2181
zookeeper.connection.timeout.ms=18000
group.initial.rebalance.delay.ms=0

2.3 启动

• nohup启动（不推荐）

nohup /opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/server.properties 1>/dev/null 2>&1 &

• 以demean模式启动（推荐）

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

• 开机启动

[root@kafka-03 init.d]# cat > /etc/init.d/kafka.sh << EOF
#!/bin/sh
#chkconfig: 2345 56 55
#description:kafka.sh
#####################
source /etc/profile
cd /opt/kafka/logs
/opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties
EOF
[root@kafka-01 init.d]# chmod 755 /etc/init.d/kafka.sh
[root@kafka-01 init.d]# systemctl enable kafka.sh
kafka.sh.service is not a native service, redirecting to /sbin/chkconfig.
Executing /sbin/chkconfig kafka.sh on

3. 使用密码

• 停止kafka

./kafka-server-stop.sh

3.1 配置文件

进入config目录修改如下文件

[root@kafka-01 config]# pwd
/opt/kafka/config

• server.properties

说明：listeners的地址三台服务器根据实际情况填写各自的ip地址。

listeners=SASL_PLAINTEXT://10.10.xxx.61:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN

• kafka_server_jaas.conf

KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="kafka"
    password="xxe888888"
    user_kafka="xxe888888"
    user_boe="xxe888888";
};

• kafka_client_jaas.conf
  在config目录添加kafka_client_jaas.conf

KafkaClient {
        org.apache.kafka.common.security.plain.PlainLoginModule required
        username="mooc"
        password="xxe888888";
};

3.2 修改启动脚本

进入bin目录

[root@kafka-01 bin]# pwd
/opt/kafka/bin

• kafka-server-start.sh
  修改kafka服务的用户名密码

说明：用${KAFKA_OPTS} 指定用户密码文件。注意写在执行kafka-run-class.sh 脚本前

if [ "x$KAFKA_OPTS" = "x"  ]; then
    export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka/config/kafka_server_jaas.conf"
fi
exec $base_dir/kafka-run-class.sh $EXTRA_ARGS kafka.Kafka "$@"

• kafka-console-producer.sh/ kafka-console-consumer.sh
  修改客户端的用户名密码。

说明：用${KAFKA_OPTS} 指定用户密码文件。注意写在执行kafka-run-class.sh 脚本前

if [ "x$KAFKA_OPTS" = "x"  ]; then
    export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka/config/kafka_client_jaas.conf"
fi

• 启动服务

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

4. 测试

• 创建topic

./kafka-topics.sh  --zookeeper localhost:2181 --create --topic Test  --partitions 1 --replication-factor 1

• 生产者
  打开一个终端，执行如下命令，进入console后随便输入一串数字

[root@kafka-01 bin]# ./kafka-console-producer.sh   --broker-list 10.10.xxx.62:9092 --topic Test --producer-property security.protocol=SASL_PLAINTEXT --producer-property sasl.mechanism=PLAIN
>111
>hongjian

• 消费者
  打开生产者的同时，打开一个消费者。可以看到接收到生产者的消息了。

[root@kafka-01 bin]# ./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic Test --consumer-property security.protocol=SASL_PLAINTEXT --consumer-property sasl.mechanism=PLAIN --from-beginning
111
hongjian

---