本文详细介绍了 Apache Kafka 4.1.1 的部署流程，涵盖从基础的单机版部署到高级安全认证配置。内容包括：Kafka 软件下载与初始化、基础配置、SASL/PLAIN 简单认证部署、以及更安全的 SASL/SCRAM 认证配置。无论您是初次接触 Kafka 还是需要在生产环境中加强安全防护，本指南都提供了完整的操作步骤和验证方法，帮助您快速搭建并保护您的 Kafka 服务。

一、单机版部署

本章节介绍如何在单台服务器上快速部署一个基础的 Kafka 4.1.1 实例。这是了解 Kafka 组件和运行原理的起点，适用于开发、测试或小规模生产环境。

部署说明：单机版部署是 Kafka 的最简运行模式，包含一个既是 Broker 又是 Controller 的节点。通过此模式可以快速验证 Kafka 的基本功能，为后续集群部署和安全配置奠定基础。

1.下载解压

首先需要从 Apache 官网获取 Kafka 二进制发行版。选择与 Scala 版本匹配的包（这里使用 2.13），下载后解压到指定目录，准备后续配置。

#官网下载
https://dlcdn.apache.org/kafka/4.1.1/kafka_2.13-4.1.1.tgz
#解压
[worker@ansible-111 software]$ tar xf kafka_2.13-4.1.1.tgz

在这里插入图片描述

2.修改配置文件

Kafka 的核心配置文件是 server.properties。需要根据实际部署环境调整关键参数，特别是网络监听地址、数据存储路径和副本因子设置。正确的配置是 Kafka 正常启动和对外服务的基础。

[worker@ansible-111 ~]$ cd /data/software/kafka_4.1.1/config
[worker@ansible-111 config]$ cat server.properties 

############################# Server Basics #############################
process.roles=broker,controller
node.id=1
# 【修改点1】单机模式下，控制器引导服务器指向自己。如果只有本机客户端，可保留localhost；若需外部连接，建议改为本机IP。
controller.quorum.bootstrap.servers=172.16.130.111:9093

############################# Socket Server Settings #############################
listeners=PLAINTEXT://:9092,CONTROLLER://:9093
inter.broker.listener.name=PLAINTEXT
# 【关键修改点2】必须将 localhost 改为服务器实际IP或主机名，否则外部客户端无法解析。
advertised.listeners=PLAINTEXT://172.16.130.111:9092
controller.listener.names=CONTROLLER
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT

# 线程与网络缓冲配置（可根据机器性能调整）
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600

############################# Log Basics #############################
# 【修改点3】建议更改为一个持久化、空间充足的目录，而非 /tmp
log.dirs=/data/software/kafka_4.1.1/logs
num.partitions=1
num.recovery.threads.per.data.dir=1

############################# Internal Topic Settings #############################
# 单机版，复制因子只能为1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1

############################# Log Retention Policy #############################
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000

3.初始化

Kafka 从 3.0 版本开始引入了存储工具 kafka-storage.sh，用于格式化数据目录并生成集群 ID。这是 Kafka 启动前的必要步骤，确保存储目录结构正确且具有唯一的集群标识。

#生成唯一的集群id
[root@ansible-111 kafka_4.1.1]# bin/kafka-storage.sh random-uuid
W2bpBFD7TIqgaAdkLswcfQ
#用该ID格式化存储目录（重要初始化步骤）
[root@ansible-111 kafka_4.1.1]# bin/kafka-storage.sh format --standalone -t W2bpBFD7TIqgaAdkLswcfQ -c /data/software/kafka_4.1.1/config/server.properties

4.启动与验证

启动 Kafka 服务进程，并通过查看日志和进程状态确认服务是否正常启动。成功启动后，Kafka 将开始监听配置的端口（9092 用于客户端，9093 用于控制器间通信）。

#启动
[root@ansible-111 kafka_4.1.1]#  nohup bin/kafka-server-start.sh config/server.properties >>nohup.out 2>&1 &
#查看日志是否启动成功
[root@ansible-111 kafka_4.1.1]# tail -f nohup.out
[root@ansible-111 kafka_4.1.1]# ps -ef | grep kafka

5.基础测试

通过 Kafka 自带的命令行工具进行基本功能测试，包括创建 Topic、生产消息和消费消息。这能验证 Kafka 服务是否正常工作，并为后续安全配置提供基准测试。

# 01 创建一个测试Topic
/data/software/kafka_4.1.1/bin/kafka-topics.sh \
  --create \
  --topic test \
  --partitions 1 \
  --replication-factor 1 \
  --bootstrap-server 172.16.130.111:9092
# 02 生产一些测试消息
echo "Hello, Kafka 2.1.1!" | /data/software/kafka_4.1.1/bin/kafka-console-producer.sh \
  --topic test \
  --bootstrap-server 172.16.130.111:9092

# 03 消费测试消息
/data/software/kafka_4.1.1/bin/kafka-console-consumer.sh \
  --topic test \
  --from-beginning \
  --bootstrap-server 172.16.130.111:9092 \
  --timeout-ms 5000

二、启用 SASL/PLAIN 安全认证

本章节介绍如何为单机版 Kafka 启用 SASL/PLAIN 认证。这是一种简单的用户名密码认证机制，适用于内部网络或与 SSL/TLS 结合使用，为 Kafka 服务增加基础的安全防护层。

配置说明：SASL/PLAIN 是 Kafka 支持的最简单的认证机制，通过 JAAS 配置文件管理用户凭证。配置后，所有客户端连接都需要提供正确的用户名和密码，防止未授权访问。

简单的用户名密码认证，通常与SSL结合使用，对于小公司来说，没必要搭建公司级别的Kerberos，使用它就比较合适。

1.创建 JAAS 配置文件

配置文件定义了 Kafka Broker 的认证模块和用户凭证。在 PLAIN 机制下，用户名和密码以明文形式存储在此文件中，因此需要严格控制文件权限。

[worker@ansible-111 config]$ cat kafka_server_jaas.conf 
KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    serviceName="kafka"  
    username="admin"
    password="jczz@1234"
    user_admin="jczz@1234"
    user_kafkaui="jczz@1234";
};

#重要提示：上述配置创建了两个用户：admin 和 kafkaui。请务必在生产环境中使用更强的密码。
#Broker 自身凭证，集群之间相互通信 username="admin"  password="jczz@1234"
#用户凭证列表,格式为user_<username>="<password>"

2.修改 Kafka 服务器配置

需要修改 server.properties 文件以启用 SASL 认证。关键更改包括：调整监听器协议、配置 SASL 机制以及更新安全协议映射。这些更改使 Kafka Broker 能够处理 SASL 认证的连接。

[worker@ansible-111 ~]$ cd /data/software/kafka_4.1.1/config
[worker@ansible-111 config]$ cat server.properties 
# 核心身份
process.roles=broker,controller
node.id=1
controller.quorum.bootstrap.servers=172.16.130.111:9093
                    
# ====== 关键修改点开始 监听器与安全映射 ======
# 使用标准监听器名称，并明确分离协议
listeners=SASL_PLAINTEXT://:9092,CONTROLLER://:9093
inter.broker.listener.name=SASL_PLAINTEXT
advertised.listeners=SASL_PLAINTEXT://172.16.130.111:9092
# 控制器监听器名称保持标准‘CONTROLLER’
controller.listener.names=CONTROLLER
# 清晰映射：CONTROLLER走PLAINTEXT，SASL_PLAINTEXT走SASL_PLAINTEXT
listener.security.protocol.map=CONTROLLER:PLAINTEXT,SASL_PLAINTEXT:SASL_PLAINTEXT
                   
# ====== SASL 配置 ======
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN
# ====== 关键修改点结束 监听器与安全映射 ======
                                                                                       # ====== 其余配置相同 ======
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=/data/software/kafka_4.1.1/logs
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
share.coordinator.state.topic.replication.factor=1
share.coordinator.state.topic.min.isr=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

3.启动kafka

通过环境变量 KAFKA_OPTS 指定 JAAS 配置文件路径，然后重启 Kafka 服务以应用 SASL/PLAIN 认证配置。启动脚本中包含了停止旧进程和启动新进程的完整流程。

[worker@ansible-111 kafka_4.1.1]$ cat restart.sh 
#!/bin/bash  
source /etc/profile
pid=`ps -ef|grep 'kafka_4.1.1' | grep -v grep | awk '{print $2}'`
kill -9 $pid
sleep 3s
#将认证文件加入到启动中
export KAFKA_OPTS="-Djava.security.auth.login.config=/data/software/kafka_4.1.1/config/kafka_server_jaas.conf" 
nohup bin/kafka-server-start.sh config/server.properties >>nohup.out 2>&1 &

#启动
[worker@ansible-111 kafka_4.1.1]$ sh restart.sh

在这里插入图片描述

4.创建客户端连接配置

启用认证后，客户端需要提供认证信息才能连接。创建 client-sasl.properties 配置文件，包含安全协议、SASL 机制和 JAAS 配置，供生产者和消费者使用。

[worker@ansible-111 kafka_4.1.1]$ cat config/client-sasl.properties 
# 指定安全协议
security.protocol=SASL_PLAINTEXT
# 指定SASL机制
sasl.mechanism=PLAIN
# 提供JAAS配置（用户名密码必须与JAAS文件中的某个user_*条目匹配）
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="jczz@1234";

5.连接kafka测试

使用配置了认证信息的客户端连接 Kafka，执行创建 Topic、列出 Topic、生产和消费消息等操作，验证 SASL/PLAIN 认证是否正常工作。

#01 创建topic
/data/software/kafka_4.1.1/bin/kafka-topics.sh \
  --create \
  --topic test-sasl-topic \
  --partitions 1 \
  --replication-factor 1 \
  --bootstrap-server 172.16.130.111:9092\
  --command-config config/client-sasl.properties
  
# 02. 列出所有Topic (测试连接)
bin/kafka-topics.sh --list \
    --bootstrap-server 172.16.130.111:9092 \
    --command-config config/client-sasl.properties

# 03. 生产消息
echo "Hello, Secure Kafka" | bin/kafka-console-producer.sh \
    --topic test-sasl-topic \
    --bootstrap-server 172.16.130.111:9092 \
    --producer.config config/client-sasl.properties

# 04. 消费消息
bin/kafka-console-consumer.sh \
    --topic test-sasl-topic \
    --from-beginning \
    --bootstrap-server 172.16.130.111:9092 \
    --consumer.config config/client-sasl.properties \
    --timeout-ms 5000

在这里插入图片描述

三、启用SASL/SCRAM认证

本章节介绍如何将认证机制从简单的 SASL/PLAIN 升级为更安全、支持动态用户管理的 SASL/SCRAM。SCRAM 解决了 PLAIN 机制密码明文传输和存储的问题，是构建更健壮安全体系的重要步骤。

升级说明：Salted Challenge Response Authentication Mechanism (SCRAM) 通过哈希和盐值保护密码，避免明文传输。用户凭证存储在 Kafka 内部 Topic 中，支持动态管理，无需重启 Broker 即可添加/删除用户。

PLAIN的加强版本，支持动态的用户增减。

1.创建 SCRAM 用户凭证

使用 Kafka 的 kafka-configs.sh 工具创建 SCRAM 用户。与 PLAIN 不同，SCRAM 用户凭证存储在 Kafka 的内部 Topic 中，而不是静态文件。此步骤为后续的 SCRAM 认证准备用户数据。

# 创建 admin 用户 (SCRAM-SHA-512)
bin/kafka-configs.sh --bootstrap-server 172.16.130.111:9092 \
  --entity-type users --entity-name admin \
  --alter --add-config 'SCRAM-SHA-512=[password=jczz@1213]'\
   --command-config config/client-sasl.properties
  # 请将 `admin-scram-pass` 替换为高强度密码

# 创建 kafkaui 用户
bin/kafka-configs.sh --bootstrap-server 172.16.130.111:9092 \
  --entity-type users --entity-name kafkaui \
  --alter --add-config 'SCRAM-SHA-512=[password=jczz@1213]' \
   --command-config config/client-sasl.properties
  # 请将 `kafkaui-scram-pass` 替换为高强度密码

# 验证用户已添加
bin/kafka-configs.sh --bootstrap-server 172.16.130.111:9092 \
  --entity-type users --describe \
  --command-config config/client-sasl.properties

#输出应显示：SCRAM-SHA-512 配置已附加到对应用户

在这里插入图片描述

2.修改 JAAS 配置文件

更新 JAAS 配置文件，将 PlainLoginModule 替换为 ScramLoginModule。SCRAM 模式下，JAAS 文件仅包含 Broker 自身的认证信息（用于启动和内部通信），用户凭证不再在此文件中定义。

重要变化：

  • PlainLoginModule 改为 ScramLoginModule。
  • 移除了 serviceName 和 user_* 条目。
  • 只保留一个用于启动Broker进程自身身份认证的 username 和 password。这里我们使用 admin 用户及其SCRAM密码。
[worker@ansible-111 kafka_4.1.1]$  cat > config/kafka_server_jaas.conf << 'EOF'
KafkaServer {
    org.apache.kafka.common.security.scram.ScramLoginModule required
    username="admin"
    password="jczz@1213"; // 使用上一步为admin用户设置的SCRAM密码
};
EOF

3.修改 server.properties 配置文件

在 server.properties 中将 SASL 机制从 PLAIN 改为 SCRAM-SHA-512（或 SCRAM-SHA-256）。这告诉 Kafka Broker 使用 SCRAM 机制处理认证请求。

[worker@ansible-111 kafka_4.1.1]$ vim config/server.properties
# 将 SASL 机制从 PLAIN 改为 SCRAM-SHA-512 (或 SCRAM-SHA-256)
sasl.enabled.mechanisms=SCRAM-SHA-512
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512
# inter.broker.listener.name 保持不变，仍为 SASL_PLAINTEXT
# 但协议本身已通过上述机制改变

注意：SCRAM-SHA-512 比 SCRAM-SHA-256 更安全，是推荐选择。请确保 listener.security.protocol.map 中 SASL_PLAINTEXT 对应的机制能支持SCRAM（默认支持）。

4.重启kafka服务

修改配置后，需要重启 Kafka 服务以使 SCRAM 认证生效。使用之前的重启脚本，确保 JAAS 配置文件被正确加载。

[worker@ansible-111 kafka_4.1.1]$ sh restart.sh 
[worker@ansible-111 kafka_4.1.1]$ tail -f nohup.out

5.客户端连接测试

创建针对 SCRAM 的客户端配置文件，使用 SCRAM 机制和对应的用户凭证连接 Kafka，执行基本的 Topic 操作以验证 SCRAM 认证是否正常工作。

#添加客户端连接配置信息
[worker@ansible-111 kafka_4.1.1]$ cat config/client-scram.properties 
security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
    username="admin" \
    password="jczz@1213";

# 1. 测试列出Topic (最基本的连接和认证测试)
bin/kafka-topics.sh --list \
    --bootstrap-server 172.16.130.111:9092 \
    --command-config config/client-scram.properties

# 2. 生产一条测试消息
echo "Test message via SCRAM" | bin/kafka-console-producer.sh \
    --topic test-topic \
    --bootstrap-server 172.16.130.111:9092 \
    --producer.config config/client-scram.properties

# 3. 消费测试消息
bin/kafka-console-consumer.sh \
    --topic test-topic \
    --from-beginning \
    --bootstrap-server 172.16.130.111:9092 \
    --consumer.config config/client-scram.properties \
    --timeout-ms 5000

在这里插入图片描述

6.添加新用户

演示 SCRAM 的动态用户管理能力。使用 kafka-configs.sh 工具可以随时添加新用户，无需修改配置文件或重启 Broker。这为生产环境中的用户权限管理提供了极大便利。

 bin/kafka-configs.sh --bootstrap-server 172.16.130.111:9092 \
  --entity-type users --entity-name kafka-canal \
  --alter --add-config 'SCRAM-SHA-512=[password=jczz@1213]'\
  --command-config config/client-scram.properties
  # 请设置高强度密码
  
#验证用户
bin/kafka-configs.sh --bootstrap-server 172.16.130.111:9092 \
  --entity-type users --describe \
  --command-config config/client-scram.properties

在这里插入图片描述

四、总结

本指南系统性地演示了 Apache Kafka 4.1.1 从基础单机部署到启用两种不同安全级别认证（SASL/PLAIN 和 SASL/SCRAM）的全过程。关键步骤包括：修改核心配置文件 server.properties 以正确绑定网络和设置监听器；使用 kafka-storage.sh 初始化存储目录；以及通过 JAAS 配置文件和安全机制参数启用认证。

核心要点：

  • 网络配置是基础：务必正确设置 advertised.listeners，否则外部客户端将无法连接。
  • 安全演进路径：从无认证，到简单的 SASL/PLAIN，再到更安全、支持动态用户管理的 SASL/SCRAM，您可以根据实际安全需求选择。
  • 用户管理：SASL/SCRAM 允许您通过 Kafka 命令行工具动态创建和管理用户，无需重启 Broker，这为生产环境用户管理提供了极大的灵活性。
  • 客户端配置：启用安全认证后，所有客户端（生产者、消费者、管理工具）都必须提供相应的认证配置文件（如 client-sasl.properties 或 client-scram.properties）才能成功连接。

遵循本指南，您不仅可以搭建一个可用的 Kafka 服务，还能为其赋予基本或增强的安全防护能力，为集成更复杂的生态系统（如 Kafka Connect, Kafka Streams）或对接外部监控、管理界面（如 Kafka UI）做好准备。请务必在生产环境中使用强密码，并考虑结合 SSL/TLS 加密传输层以实现全方位保护。

# kafka # 安全 # linq
Logo
2048 AI社区

有“AI”的1024 = 2048，欢迎大家加入2048 AI社区

更多推荐

Lighthouse安全组自动化审计与加固：基于MCP协议的智能运维实践

本文介绍了一个简易版网络资源监控器的Rust实现方案。该系统通过读取配置文件中的URL列表，定期发送HTTP请求来监测网站可用性，记录响应时间和状态码，并在网站不可达时发出告警。系统采用reqwest库进行异步HTTP请求，tokio处理异步运行时，clap解析命令行参数，csv库存储监控结果到文件。实现过程包括：1)单个网站监测函数；2)读取监控列表；3)异步定时监控模块；4)CSV日志记录功能

avatar 2048 AI社区
cover

前后端分离校园资产管理系统｜SpringBoot+Vue+MyBatis+MySQL完整源码+部署教程

avatar 2048 AI社区

Spring Bean介绍（被Spring IoC容器创建、装配、管理的Java对象）IoC容器、控制反转（IoC）、依赖注入（DI）、循环依赖

Spring Bean = 被Spring IoC容器创建、装配、管理的Java对象。IoC容器（Inversion of Control Container）是Spring框架的核心引擎，负责创建、组装、管理所有Bean，并将“控制权”从开发者手中“反转”交给框架。它不一定是“JavaBean”（无参构造+getter/setter），而是一个生命周期由Spring全权负责的对象。你可以把Spr

avatar 2048 AI社区