软件篇-kafka(五)-集群-kafka集群

1.前置

紧跟上文,还是选择相同的三台机器，操作系统为centos,且三台机器的ip如下：

  • 192.168.80.130
  • 192.168.80.131
  • 192.168.80.132

同时需要将kafka上传到这三台机器上，安装包跟之前讲解单机版安装版本一样

2.安装

  • 解压

    tar -zxvf kafka_2.12-2.8.0.tgz -C /opt/

mv /opt/kafka_2.12-2.8.0/ /opt/kafka

chown -R kafka:kafka /opt/kafka

    注意：三台机器都要操作

  • 创建数据目录，用于kafka存储数据

    mkdir -p /var/data/kafka

chown -R kafka:kafka /var/data/kafka

    注意：三台机器都要操作

  • 配置

    修改配置文件，vim /opt/kafka/config/server.properties

    # broker id 集群唯一
broker.id=130

listeners=PLAINTEXT://192.168.80.130:9092
advertised.listeners=PLAINTEXT://192.168.80.130:9092


num.network.threads=3

num.io.threads=8

socket.send.buffer.bytes=102400


socket.request.max.bytes=104857600

log.dirs=/var/data/kafka

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

# zookeeper集群地址
zookeeper.connect=192.168.80.130:2181,192.168.80.131:2181,192.168.80.132:2181

# Timeout in ms for connecting to zookeeper
zookeeper.connection.timeout.ms=18000

group.initial.rebalance.delay.ms=0

    配置文件的内容在之前单机版时已经详细解释过了，这里不再解释

    这是在192.168.80.130机器上的配置,其他机器上也要进行相对应的配置

  • 防火墙

    关闭防火墙或者将端口加入到防火墙里面

    systemctl stop  firewalld

firewall-cmd --add-port 9092/tcp --permanent
firewall-cmd --reload

    注意: 三台机器都要操作

  • 配置环境变量

    vim /etc/profile,在文件末尾追加一下内容

    export KAFKA_HOME=/opt/kafka
export PATH=$PATH:$KAFKA_HOME/bin

    让配置生效

    source /etc/profile

    注意：三台都要配置

  • 服务

    在/etc/systemd/system新建kafka.service文件

    touch kafka.service
chmod u+x,o+x,g+x kafka.service

    文件内容如下：

    [Unit]
Description=kafka
Requires=network.target remote-fs.target
After=network.target remote-fs.target zookeeper.service
 
[Service]
Type=forking
User=kafka
Group=kafka
Environment=JAVA_HOME=/opt/jdk
ExecStart=/opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties
ExecStop=/opt/kafka/bin/kafka-server-stop.sh
# 启动失败后重启
Restart=on-failure
# 每次尝试重启间隔60秒
StartLimitInterval=60
# 最终尝试重启3000次
StartLimitBurst=3000
[Install]
WantedBy=multi-user.target

    注意三台机器都需要操作

  • 命令

    # 配置生效
systemctl daemon-reload

# 启动
systemctl start kafka

# 查看启动状态
systemctl status kafka

# 查看启动日志
journalctl -xe

# 关闭
systemctl stop kafka

# 开启自启动
systemctl enable kafka

3. 监控

kafka集群安装了以后，为了更加方便的管理集群，安装kafka-eagle,对集群信息一目了然，安装细节之前的章节已经介绍了，这里再去安装一次

  • 开启JMX监控

    编辑vim /opt/kafka/bin/kafka-server-start.sh

    在此处添加以下代码

    export JMX_PORT="9999"
    image-20210512112200223

    编辑vim /opt/kafka/bin/kafka-run-class.sh

    将此处代码修改为以下内容

    if [ -z "$KAFKA_JMX_OPTS" ]; then
  KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote.rmi.port=$JMX_PORT -Djava.rmi.server.hostname=192.168.80.110 -Dcom.sun.anagement.jmxremote.local.only=false   -Dcom.sun.management.jmxremote.authenticate=false  -Dcom.sun.management.jmxremote.ssl=false "
fi
    image-20210511142335936

    重启kafka

    systemctl restart kafka

    注意：三台机器都要配置

  • 安装kafka-eagle

    在一台机器上安装kafka-eagle即可，不需要在所以的机器安装

    笔者这里是在192.168.80.130机器上安装

    解压

    tar xvf kafka-eagle-bin-2.0.0.tar.gz  -C .

cd kafka-eagle-bin-2.0.0
 
tar -xvf kafka-eagle-web-2.0.0-bin.tar.gz -C /opt/

mv /opt/kafka-eagle-web-2.0.0 /opt/kafka-eagle

chown -R kafka:kafka /opt/kafka-eagle

    修改配置文件 vim /opt/kafka-eagle/conf/system-config.properties

    内容如下：

    ######################################
# multi zookeeper & kafka cluster list
######################################
kafka.eagle.zk.cluster.alias=cluster1
cluster1.zk.list=192.168.80.130:2181,192.168.80.131:2181,192.168.80.132:2181

######################################
# zookeeper enable acl
######################################
cluster1.zk.acl.enable=false

######################################
# broker size online list
######################################
cluster1.kafka.eagle.broker.size=20

######################################
# zk client thread limit
######################################
kafka.zk.limit.size=25

######################################
# kafka eagle webui port
######################################
kafka.eagle.webui.port=8048

######################################
# kafka offset storage
######################################
cluster1.kafka.eagle.offset.storage=kafka

######################################
# kafka metrics, 15 days by default
######################################
kafka.eagle.metrics.charts=true
kafka.eagle.metrics.retain=15


######################################
# kafka sql topic records max
######################################
kafka.eagle.sql.topic.records.max=5000
kafka.eagle.sql.fix.error=true

######################################
# delete kafka topic token
######################################
kafka.eagle.topic.token=keadmin

######################################
# kafka sasl authenticate
######################################
cluster1.kafka.eagle.sasl.enable=false

######################################
# kafka ssl authenticate
######################################
cluster1.kafka.eagle.ssl.enable=false

######################################
# kafka sqlite jdbc driver address
######################################
kafka.eagle.driver=org.sqlite.JDBC
kafka.eagle.url=jdbc:sqlite:/opt/kafka-eagle/db/ke.db
kafka.eagle.username=root
kafka.eagle.password=www.kafka-eagle.org

    配置环境变量vim /etc/profile

    一台机器配置即可

    export KE_HOME=/opt/kafka-eagle
export PATH=$PATH:$KE_HOME/bin

    让配置生效

    source /etc/profile

    将端口添加到防火墙或者关闭防火墙

    # 关闭防火墙
systemctl stop firewalld

# 添加端口
firewall-cmd --add-port=8048/tcp --permanent
firewall-cmd --reload

    创建服务文件vim /etc/systemd/system/kafka-eagle.service，内容如下：

    [Unit]
Description=Kafka Eagle
After=kafka.service

[Service]
Environment=KE_HOME=/opt/kafka-eagle
Environment=JAVA_HOME=/opt/jdk
User=kafka
Group=kafka
Type=forking

ExecStart=/opt/kafka-eagle/bin/ke.sh start
ExecReload=/opt/kafka-eagle/bin/ke.sh restart
ExecStop=/opt/kafka-eagle/bin/ke.sh stop

#启动失败后重启
Restart=on-failure
#每次尝试重启间隔60秒
StartLimitInterval=60
#最终尝试重启50次
StartLimitBurst=50

[Install]
WantedBy=multi-user.target

    kafka-eagle命令操作如下

    # 配置生效
systemctl daemon-reload

# 启动
systemctl start kafka-eagle

# 查看启动状态
systemctl status kafka-eagle

# 查看启动日志
journalctl -xe

# 关闭
systemctl stop kafka-eagle

# 开启自启动
systemctl enable kafka-eagle

    启动成功后，访问地址192.168.80.130:8048地址，即可看到监控网页

    image-20210512115149301

    账号：admin/123456

    image-20210512115534065

    也可以通过大屏展示监控kafka

    [图片上传失败...(image-53c984-1620823151774)]

    例如我创建了一个test_topic主题，且分区数为3，副本数为3

    image-20210512134113063

    查看该主题即可看到分区分布信息，如下：

    image-20210512134208188

    自此集群搭建完成

4. 安装(docker)

创建目录用来进行数据挂载

mkdir -p /root/kafka/data && chmod 777 /root/kafka/data
mkdir -p /root/kafka/config

当zookeeper集群创建好了以后就可以去创建kafka集群了，具体步骤如下：

在192.168.80.130机器上输入以下指令

docker run -di --network=host --privileged=true \
--name kafka -p 9092:9092 \
-e KAFKA_BROKER_ID=1 \
-e KAFKA_ZOOKEEPER_CONNECT=192.168.80.130:2181,192.168.80.131:2181,192.168.80.132:2181 \
-e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.80.130:9092 \
-e ALLOW_PLAINTEXT_LISTENER=yes \
-v /root/kafka/data:/bitnami/kafka/data \
-v /root/kafka/config:/bitnami/kafka/config \
bitnami/kafka:2.6.0

在192.168.80.131机器上输入以下指令

docker run -di --network=host --privileged=true \
--name kafka -p 9092:9092 \
-e KAFKA_BROKER_ID=2 \
-e KAFKA_ZOOKEEPER_CONNECT=192.168.80.130:2181,192.168.80.131:2181,192.168.80.132:2181 \
-e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.80.131:9092 \
-e ALLOW_PLAINTEXT_LISTENER=yes \
-v /root/kafka/data:/bitnami/kafka/data \
-v /root/kafka/config:/bitnami/kafka/config \
bitnami/kafka:2.6.0

在192.168.80.132机器上输入以下指令

docker run -di --network=host --privileged=true \
--name kafka -p 9092:9092 \
-e KAFKA_BROKER_ID=3 \
-e KAFKA_ZOOKEEPER_CONNECT=192.168.80.130:2181,192.168.80.131:2181,192.168.80.132:2181 \
-e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.80.132:9092 \
-e ALLOW_PLAINTEXT_LISTENER=yes \
-v /root/kafka/data:/bitnami/kafka/data \
-v /root/kafka/config:/bitnami/kafka/config \
bitnami/kafka:2.6.0
最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
平台声明：文章内容（如有图片或视频亦包括在内）由作者上传并发布，文章内容仅代表作者本人观点，简书系信息发布平台，仅提供信息存储服务。

推荐阅读更多精彩内容