环境

主机名 IP 操作系统 kafka 版本
kafka224 192.168.1.224 CentOS7.5 2.12
kafka225 192.168.1.225 CentOS7.5 2.12
kafka226 192.168.1.226 CentOS7.5 2.12
  • 下载 kafka_2.12-2.1.0.tgz
  • zookeeper 连接: 192.168.1.221:2181,192.168.1.222:2181,192.168.1.223:2181

各节点初始配置

  • 关闭 selinux、防火墙
  • 部署 java 运行环境
  • 创建数据目录 
    1

    mkdir -p /var/lib/kafka
  • 创建日志目录 
    1

    mkdir -p /var/log/kafka

部署 kafka

  • 登陆 kafka224，下载 kafka，解压至 /opt/ 下
  • 修改日志目录 
    1
2

    cd /opt/kafka/
rm -rf logs && ln -s /var/log/kafka logs
  • 修改 /opt/kafka/config/server.properties 
    1
2
3
4

    broker.id=224
listeners=PLAINTEXT://192.168.1.224:9092
log.dirs=/var/lib/kafka
zookeeper.connect=192.168.1.221:2181,192.168.1.222:2181,192.168.1.223:2181
  • 启动脚本(/opt/kafka/bin/kafka-server-start.sh)太繁琐，我精简了下 
     1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

    #!/bin/bash
base_dir=$(dirname $0)
if [ "x$KAFKA_LOG4J_OPTS" = "x" ]; then
    export KAFKA_LOG4J_OPTS="-Dlog4j.configuration=file:$base_dir/../config/log4j.properties"
fi
if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
    # jvm 堆内存根据实际情况修改
    export KAFKA_HEAP_OPTS="-Xmx2G -Xms2G"
    # 开启 jmx
    export JMX_PORT=1099
fi
EXTRA_ARGS=${EXTRA_ARGS-'-name kafkaServer -loggc'}
EXTRA_ARGS="-daemon "$EXTRA_ARGS
exec $base_dir/kafka-run-class.sh $EXTRA_ARGS kafka.Kafka "$base_dir/../config/server.properties"
  • 打包 kafka 目录，复制到 kafka225 和 kafka226 上，并修改 server.properties 
    1
2
3
4
5
6

    # kafka225
broker.id=225
listeners=PLAINTEXT://192.168.1.225:9092
# kafka226
broker.id=226
listeners=PLAINTEXT://192.168.1.226:9092

启动集群(两种启动方式)

  • 直接启动二进制
    • 在每个节点上启动 kafka 服务 
      1

      /opt/kafka/bin/kafka-server-start.sh
  • systemd 启动
    • 创建 /usr/lib/systemd/system/kafka.service 
       1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

      [Unit]
Description=Kafka
Requires=network.service
After=network.service
[Service]
Environment=JAVA_HOME=/opt/jre
ExecStart=/opt/kafka/bin/kafka-server-start.sh
ExecStop=/opt/kafka/bin/kafka-server-stop.sh
Type=forking
KillMode=none
[Install]
WantedBy=multi-user.target
    • 启动 kafka 服务 
      1
2

      systemctl daemon-reload
systemctl start kafka