kafka单节点安装

Kafka 2.8.0 版本移除了对Zookeeper的依赖，该版本将依赖于 ZooKeeper 的控制器改造成了基于 Kafka Raft 的 Quorm 控制器。

安装不同版本

安装 kafka 3.5.0

  • 1.安装 jdk 1.8.0_391
curl -sL http://iso.sqlfans.cn/jdk/install_jdk_8u391.sh | bash
source /etc/profile
java -version
  • 2.下载 kafka 3.5.0
mkdir -p /data
curl -sL http://iso.sqlfans.cn/linux/kafka_2.12-3.5.0.tgz -o /opt/kafka_2.12-3.5.0.tgz
tar -xzf /opt/kafka_2.12-3.5.0.tgz -C /opt/
mv /opt/kafka_2.12-3.5.0 /data/kafka_9092
  • 3.配置 server.properties
cp /data/kafka_9092/config/kraft/server.properties /data/kafka_9092/config/kraft/server.properties.bak.$(date +%Y%m%d%H%M%S)
sed -i 's/^process.roles=.*/process.roles=broker,controller/' /data/kafka_9092/config/kraft/server.properties
sed -i 's/^controller.quorum.voters=.*/controller.quorum.voters=1@192.168.31.101:9093/' /data/kafka_9092/config/kraft/server.properties
sed -i 's/^log.dirs=.*/log.dirs=\/data\/kafka_9092\/logs/' /data/kafka_9092/config/kraft/server.properties

sed -i 's/^node.id=.*/node.id=1/' /data/kafka_9092/config/kraft/server.properties
sed -i 's/^listeners=.*/listeners=PLAINTEXT:\/\/192.168.31.101:9092,CONTROLLER:\/\/192.168.31.101:9093/' /data/kafka_9092/config/kraft/server.properties
sed -i 's/^advertised.listeners=.*/advertised.listeners=PLAINTEXT:\/\/192.168.31.101:9092/' /data/kafka_9092/config/kraft/server.properties
cat /data/kafka_9092/config/kraft/server.properties | grep -Ev '^$|#' | egrep "(process.|node.id|controller.|listeners|log.dir|sadvertised.)"
  • 4.启动进程
KAFKA_CLUSTER_ID="$(/data/kafka_9092/bin/kafka-storage.sh random-uuid)"
echo "集群ID: $KAFKA_CLUSTER_ID"
/data/kafka_9092/bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID -c /data/kafka_9092/config/kraft/server.properties
nohup /data/kafka_9092/bin/kafka-server-start.sh /data/kafka_9092/config/kraft/server.properties > /data/kafka_9092/logs/kafka_startup.log 2>&1 &
sleep 5 && netstat -lnpt | egrep "(9092|9093)"

安装 kafka 4.3.0

  • 1.安装 jdk 25.0.1

注：kafka 4.3.0 必须运行在 Java 17 或更高版本上

curl -sL http://iso.sqlfans.cn/jdk/install_jdk_2501.sh | bash
source /etc/profile
java -version
  • 2.下载 kafka 4.3.0
mkdir -p /data
wget -c https://archive.apache.org/dist/kafka/4.3.0/kafka_2.13-4.3.0.tgz
tar -xzf /opt/kafka_2.13-4.3.0.tgz -C /opt/
mv /opt/kafka_2.13-4.3.0 /data/kafka_9092
  • 3.配置 server.properties
cp /data/kafka_9092/config/server.properties /data/kafka_9092/config/server.properties.bak.$(date +%Y%m%d%H%M%S)
sed -i 's/^process.roles=.*/process.roles=broker,controller/' /data/kafka_9092/config/server.properties
sed -i 's/^controller.quorum.voters=.*/controller.quorum.voters=1@192.168.31.101:9093/' /data/kafka_9092/config/server.properties
sed -i 's/^log.dirs=.*/log.dirs=\/data\/kafka_9092\/logs/' /data/kafka_9092/config/server.properties

sed -i 's/^node.id=.*/node.id=1/' /data/kafka_9092/config/server.properties
sed -i 's/^listeners=.*/listeners=PLAINTEXT:\/\/192.168.31.101:9092,CONTROLLER:\/\/192.168.31.101:9093/' /data/kafka_9092/config/server.properties
sed -i 's/^advertised.listeners=.*/advertised.listeners=PLAINTEXT:\/\/192.168.31.101:9092/' /data/kafka_9092/config/server.properties
cat /data/kafka_9092/config/server.properties | grep -Ev '^$|#' | egrep "(process.|node.id|controller.|listeners|log.dir|sadvertised.)"
  • 4.启动进程

注：相较 3.5.0 版本，格式化Kafka存储目录的过程需要添加参数 --standalone

KAFKA_CLUSTER_ID="$(/data/kafka_9092/bin/kafka-storage.sh random-uuid)"
echo "集群ID: $KAFKA_CLUSTER_ID"
/data/kafka_9092/bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID -c /data/kafka_9092/config/server.properties --standalone
nohup /data/kafka_9092/bin/kafka-server-start.sh /data/kafka_9092/config/server.properties > /data/kafka_9092/kafka_startup.log 2>&1 &
sleep 5 && netstat -lnpt | egrep "(9092|9093)"

附录

查看消息队列的延迟

  • 创建 topic 主题
# 创建topic：指定broker及1个副本及3个partition
/data/kafka_9092/bin/kafka-topics.sh --bootstrap-server 192.168.31.101:9092 --create --replication-factor 1 --partitions 3 --topic mytopic
/data/kafka_9092/bin/kafka-topics.sh --bootstrap-server 192.168.31.101:9092 --list
/data/kafka_9092/bin/kafka-topics.sh --bootstrap-server 192.168.31.101:9092 --describe --topic mytopic

# 生产消息：写入2条测试消息
echo "message123" | /data/kafka_9092/bin/kafka-console-producer.sh --bootstrap-server 192.168.31.101:9092 --topic mytopic
echo "message456" | /data/kafka_9092/bin/kafka-console-producer.sh --bootstrap-server 192.168.31.101:9092 --topic mytopic

# 删除topic
# /data/kafka_9092/bin/kafka-topics.sh --bootstrap-server 192.168.31.101:9092 --delete --topic mytopic
  • 查看特定消费组的延迟
# 消费消息：设定从头消费，执行 Ctrl-C 随时停止
/data/kafka_9092/bin/kafka-console-consumer.sh --bootstrap-server 192.168.31.101:9092 --topic mytopic --group mygroup --from-beginning

# 列出所有消费者组
/data/kafka_9092/bin/kafka-consumer-groups.sh --bootstrap-server 192.168.31.101:9092 --list

# 查看特定消费组的延迟
/data/kafka_9092/bin/kafka-consumer-groups.sh --bootstrap-server 192.168.31.101:9092 --describe --group mygroup
关键指标 指标的含义 备注
LAG 消费者与消息最新生产进度之间的消费延迟 若该值不为0或很大，说明消费者处理速度跟不上生产速度
CURRENT-OFFSET 当前消费的位置 -
LOG-END-OFFSET 分区最新的消息位置 -
  • LAG 可以直观的查看延迟，若 CURRENT-OFFSET 与 LOG-END-OFFSET 两者相等表示没有积压

注：下图是某生产环境的 kafka 延迟情况：

开启幂等性保证生产者消息不重复

  • 开启幂等性后，Kafka 会为每个生产者分配一个唯一的 PID (Producer ID)。生产者发送的每条消息（或消息批次）都会带上一个从0开始递增的序列号，能有效避免因网络抖动等异常导致的消息重复问题。
  • 如何开启与配置：在代码中配置起来很简单，核心就是设置 enable.idempotence 为 true，如下以 java 为例：
Properties props = new Properties();
props.put("enable.idempotence", "true"); // 开启幂等性
props.put("acks", "all");   // 要求消息被所有 ISR 副本确认，这是幂等性的前提条件之一
props.put("max.in.flight.requests.per.connection", "5");  // 控制生产者在收到响应前，最多能发送多少个未确认的请求

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

如何彻底卸载kafka

cd /opt/
ps aux | grep kafka | grep -v grep | awk '{print $2}' | xargs kill -9 2> /dev/null
rm -rf /data/kafka_9092
Copyright © www.sqlfans.cn 2021 - 2026 All Right Reserved更新时间： 2026-08-17 10:48:00

results matching ""

    No results matching ""