#!/bin/bash# Modify the link if you want to download other versionKAFKA_DOWNLOAD_URL="https://dlcdn.apache.org/kafka/3.1.0/kafka_2.13-3.1.0.tgz"# Please use your own server ipSERVERS=("192.168.1.1" "192.168.1.2" "192.168.1.3")ID=0MECHINE_IP=$(hostname -i)echo "Mechine IP: "${MECHINE_IP}LENGTH=${#SERVERS[@]}for (( i=0; i<${LENGTH}; i++ ));do if [ "${SERVERS[$i]}" = "${MECHINE_IP}" ]; then ID=$((i+1)) fidoneecho "ID: "${ID}if [ "${ID}" -eq "0" ]; then echo "Mechine IP is not matched to server list" exit 1fiZOOKEEPER_CONNECT=$(printf ",%s:2181" "${SERVERS[@]}")ZOOKEEPER_CONNECT=${ZOOKEEPER_CONNECT:1}echo "Zookeeper Connect: "${ZOOKEEPER_CONNECT}echo "---------- Update yum ----------"yum update -yyum install -y wgetecho "---------- Install java ----------"yum -y install java-1.8.0-openjdkjava -versionecho "---------- Create kafka user & group ----------"groupadd -r kafkauseradd -g kafka -r kafka -s /bin/falseecho "---------- Download kafka ----------"cd /optwget ${KAFKA_DOWNLOAD_URL} -O kafka.tgzmkdir -p kafkatar -xzf kafka.tgz -C kafka --strip-components=1chown -R kafka:kafka /opt/kafkaecho "---------- Install and start zookeeper ----------"mkdir -p /data/zookeeperchown -R kafka:kafka /data/zookeeperecho "${ID}" > /data/zookeeper/myid# zookeeper config# https://zookeeper.apache.org/doc/r3.1.2/zookeeperAdmin.html#sc_configurationcat <<EOF > /opt/kafka/config/zookeeper-cluster.properties# the directory where the snapshot is stored.dataDir=/data/zookeeper# the port at which the clients will connectclientPort=2181# setting number of connections to unlimitedmaxClientCnxns=0# keeps a heartbeat of zookeeper in millisecondstickTime=2000# time for initial synchronizationinitLimit=10# how many ticks can pass before timeoutsyncLimit=5# define servers ip and internal ports to zookeeperEOFfor (( i=0; i<${LENGTH}; i++ ));do INDEX=$((i+1)) echo "server.${INDEX}=${SERVERS[$i]}:2888:3888" >> /opt/kafka/config/zookeeper-cluster.propertiesdone# zookeeper.servicecat <<EOF > /usr/lib/systemd/system/zookeeper.service[Unit]Description=Apache Zookeeper server (Kafka)Documentation=http://zookeeper.apache.orgRequires=network.target remote-fs.targetAfter=network.target remote-fs.target[Service]Type=simpleUser=kafkaGroup=kafkaExecStart=/opt/kafka/bin/zookeeper-server-start.sh /opt/kafka/config/zookeeper-cluster.propertiesExecStop=/opt/kafka/bin/zookeeper-server-stop.shRestart=on-failure[Install]WantedBy=multi-user.targetEOFsystemctl daemon-reloadsystemctl start zookeeper && systemctl enable zookeeperecho "---------- Install and start kafka ----------"mkdir -p /data/kafkachown -R kafka:kafka /data/kafka# kafka config# https://kafka.apache.org/documentation/#configurationcat <<EOF > /opt/kafka/config/server-cluster.properties# The id of the broker. This must be set to a unique integer for each broker.broker.id=${ID}# Hostname and port the broker will advertise to producers and consumers. If not set,# it uses the value for "listeners" if configured. Otherwise, it will use the value# returned from java.net.InetAddress.getCanonicalHostName().advertised.listeners=PLAINTEXT://${MECHINE_IP}:9092# A comma separated list of directories under which to store log fileslog.dirs=/data/kafka# The default number of log partitions per topic. More partitions allow greater# parallelism for consumption, but this will also result in more files across# the brokers.num.partitions=1# The replication factor for the group metadata internal topics "__consumer_offsets" and "__transaction_state"# For anything other than development testing, a value greater than 1 is recommended to ensure availability such as 3.offsets.topic.replication.factor=1transaction.state.log.replication.factor=1transaction.state.log.min.isr=1# The minimum age of a log file to be eligible for deletion due to agelog.retention.hours=168# The maximum size of a log segment file. When this size is reached a new log segment will be created.log.segment.bytes=1073741824# The interval at which log segments are checked to see if they can be deleted according# to the retention policieslog.retention.check.interval.ms=300000# Zookeeper connection string (see zookeeper docs for details).# This is a comma separated host:port pairs, each corresponding to a zk# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".# You can also append an optional chroot string to the urls to specify the# root directory for all kafka znodes.zookeeper.connect=${ZOOKEEPER_CONNECT}/kafka# Timeout in ms for connecting to zookeeperzookeeper.connection.timeout.ms=60000EOF# kafka.servicecat <<EOF > /usr/lib/systemd/system/kafka.service[Unit]Description=Apache Kafka server (broker)Documentation=http://kafka.apache.org/documentation.htmlRequires=network.target remote-fs.targetAfter=network.target remote-fs.target kafka-zookeeper.service [Service]Type=simpleUser=kafkaGroup=kafkaExecStart=/opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/server-cluster.propertiesExecStop=/opt/kafka/bin/kafka-server-stop.shRestart=on-failure [Install]WantedBy=multi-user.targetEOFsystemctl daemon-reloadsystemctl start kafka && systemctl enable kafka
# 启动 zookeepersystemctl start zookeeper# 停止 zookeepersystemctl stop zookeeper# 重启 zookeepersystemctl restart zookeeper# 查看 zookeeper 日志systemctl status zookeeper -l# 启动 kafkasystemctl start kafka# 停止 kafkasystemctl stop kafka# 重启 kafkasystemctl restart kafka# 查看 kafka 日志systemctl status kafka -l
# 进入 kafka bin 目录cd /opt/kafka/bin/# 创建一个 topickafka-topics.sh --create --topic test --partitions 3 --replication-factor 1 --bootstrap-server localhost:9092# 查看 topic 描述kafka-topics.sh --topic test --describe --bootstrap-server localhost:9092# 启动生产者然后输入消息kafka-console-producer.sh --topic test --bootstrap-server localhost:9092# 启动消费者消费消息kafka-console-consumer.sh --topic test --from-beginning --bootstrap-server localhost:9092# 删除 topickafka-topics.sh --topic test --delete --bootstrap-server localhost:9092
1. 以下代码主要指定下载 kafka 的版本以及服务器 IP 列表，可根据实际情况进行调整。
# Modify the link if you want to download other versionKAFKA_DOWNLOAD_URL="https://dlcdn.apache.org/kafka/3.1.0/kafka_2.13-3.1.0.tgz"# Please use your own server ipSERVERS=("192.168.1.1" "192.168.1.2" "192.168.1.3")
2. 以下代码主要用于生成 zookeeper id 和 kafka broker id 以及拼接 kafka 配置中的 zookeeper 连接串，通过本机 IP 与填写的 IP 列表进行匹配，如果本机 IP 等于第一个服务器 IP，则 ID为 1，等于第二个服务器 IP，则 ID为 2，等于第二个服务器 IP，则 ID为 3，以此类推；本机 IP 不在填写的 IP 列表中，则会退出安装。
ID=0MECHINE_IP=$(hostname -i)echo "Mechine IP: "${MECHINE_IP}LENGTH=${#SERVERS[@]}for (( i=0; i<${LENGTH}; i++ ));do if [ "${SERVERS[$i]}" = "${MECHINE_IP}" ]; then ID=$((i+1)) fidoneecho "ID: "${ID}if [ "${ID}" -eq "0" ]; then echo "Mechine IP is not matched to server list" exit 1fiZOOKEEPER_CONNECT=$(printf ",%s:2181" "${SERVERS[@]}")ZOOKEEPER_CONNECT=${ZOOKEEPER_CONNECT:1}
3. 更新 yum 源，并安装 wget 下载工具
yum update -yyum install -y wget
4. 安装 java 8
yum -y install java-1.8.0-openjdkjava -version
5. 创建 kafka 用户及组
groupadd -r kafkauseradd -g kafka -r kafka -s /bin/false
6. 下载并解压 kafka 可执行程序
cd /optwget ${KAFKA_DOWNLOAD_URL} -O kafka.tgzmkdir -p kafkatar -xzf kafka.tgz -C kafka --strip-components=1chown -R kafka:kafka /opt/kafka
7. 创建 zookeeper 目录，创建 zookeeper id
mkdir -p /data/zookeeperchown -R kafka:kafka /data/zookeeperecho "${ID}" > /data/zookeeper/myid
8. 生成 zookeeper 配置文件，详细说明可参考：https://zookeeper.apache.org/doc/r3.1.2/zookeeperAdmin.html#sc_configuration
cat <<EOF > /opt/kafka/config/zookeeper-cluster.properties# the directory where the snapshot is stored.dataDir=/data/zookeeper# the port at which the clients will connectclientPort=2181# setting number of connections to unlimitedmaxClientCnxns=0# keeps a heartbeat of zookeeper in millisecondstickTime=2000# time for initial synchronizationinitLimit=10# how many ticks can pass before timeoutsyncLimit=5# define servers ip and internal ports to zookeeperEOFfor (( i=0; i<${LENGTH}; i++ ));do INDEX=$((i+1)) echo "server.${INDEX}=${SERVERS[$i]}:2888:3888" >> /opt/kafka/config/zookeeper-cluster.propertiesdone
9. 创建 zookeeper systemd 管理文件，启动并设置开机启动 zookeeper
cat <<EOF > /usr/lib/systemd/system/zookeeper.service[Unit]Description=Apache Zookeeper server (Kafka)Documentation=http://zookeeper.apache.orgRequires=network.target remote-fs.targetAfter=network.target remote-fs.target[Service]Type=simpleUser=kafkaGroup=kafkaExecStart=/opt/kafka/bin/zookeeper-server-start.sh /opt/kafka/config/zookeeper-cluster.propertiesExecStop=/opt/kafka/bin/zookeeper-server-stop.shRestart=on-failure[Install]WantedBy=multi-user.targetEOFsystemctl daemon-reloadsystemctl start zookeeper && systemctl enable zookeeper
10. 创建 kafka 目录
mkdir -p /data/kafkachown -R kafka:kafka /data/kafka
11. 生成 kafka 配置文件，详细说明可参考：https://kafka.apache.org/documentation/#configuration
cat <<EOF > /opt/kafka/config/server-cluster.properties# The id of the broker. This must be set to a unique integer for each broker.broker.id=${ID}# Hostname and port the broker will advertise to producers and consumers. If not set,# it uses the value for "listeners" if configured. Otherwise, it will use the value# returned from java.net.InetAddress.getCanonicalHostName().advertised.listeners=PLAINTEXT://${MECHINE_IP}:9092# A comma separated list of directories under which to store log fileslog.dirs=/data/kafka# The default number of log partitions per topic. More partitions allow greater# parallelism for consumption, but this will also result in more files across# the brokers.num.partitions=1# The replication factor for the group metadata internal topics "__consumer_offsets" and "__transaction_state"# For anything other than development testing, a value greater than 1 is recommended to ensure availability such as 3.offsets.topic.replication.factor=1transaction.state.log.replication.factor=1transaction.state.log.min.isr=1# The minimum age of a log file to be eligible for deletion due to agelog.retention.hours=168# The maximum size of a log segment file. When this size is reached a new log segment will be created.log.segment.bytes=1073741824# The interval at which log segments are checked to see if they can be deleted according# to the retention policieslog.retention.check.interval.ms=300000# Zookeeper connection string (see zookeeper docs for details).# This is a comma separated host:port pairs, each corresponding to a zk# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".# You can also append an optional chroot string to the urls to specify the# root directory for all kafka znodes.zookeeper.connect=${ZOOKEEPER_CONNECT}/kafka# Timeout in ms for connecting to zookeeperzookeeper.connection.timeout.ms=60000EOF
12. 创建 kafka systemd 管理文件，启动并设置开机启动 kafka
cat <<EOF > /usr/lib/systemd/system/kafka.service[Unit]Description=Apache Kafka server (broker)Documentation=http://kafka.apache.org/documentation.htmlRequires=network.target remote-fs.targetAfter=network.target remote-fs.target kafka-zookeeper.service [Service]Type=simpleUser=kafkaGroup=kafkaExecStart=/opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/server-cluster.propertiesExecStop=/opt/kafka/bin/kafka-server-stop.shRestart=on-failure [Install]WantedBy=multi-user.targetEOFsystemctl daemon-reloadsystemctl start kafka && systemctl enable kafka
按照上述的操作，你将快速完成 kafka 集群安装，如有问题可以在文章留言。