# 一、环境部署


```js
准备的机器
192.168.201.112
192.168.201.113
```
都安装好java jdk
```js
rpm -ivh  jdk-21_linux-x64_bin.rpm
```

都解压压缩包

```js
tar -zxvf apache-zookeeper-3.9.4-bin.tar.gz
tar -zxvf kafka_2.12-3.6.2.tgz
```

# 二、修zookeepe改配置文件
创建存储数据目录

```js
mkdir -p /kafka/zookeeper_data
```
修改192.168.201.112、192.168.201.113机器的配置文件
```js
vi  /kafka/apache-zookeeper-3.9.4-bin/conf/zoo.cfg
```

```js
# ZooKeeper 的时间基准单位（毫秒），所有时间相关配置都以它为倍数
# 比如后续的 initLimit、syncLimit 都是 "tickTime 的倍数"
tickTime=2000

# 从节点（Follower）启动后，与主节点（Leader）同步初始数据的最大超时时间
# 计算方式：initLimit × tickTime = 10 × 2000ms = 20秒
# 超过这个时间未同步成功，从节点会被判定为连接失败
initLimit=10

# 运行中，从节点与主节点之间通信的最大超时时间（如请求响应超时）
# 计算方式：syncLimit × tickTime = 5 × 2000ms = 10秒
# 超过这个时间未收到响应，主节点会认为从节点失联
syncLimit=5

# 数据存储目录（核心！必须改）
# 存储内容：内存快照（snapshot）、事务日志（记录所有写操作，用于恢复数据）
# 注意：/tmp 是临时目录，Linux 重启后会自动清空，生产环境必须改成永久目录
# 示例：dataDir=/data/zookeeper（需提前创建该目录并授权）
dataDir=/kafka/zookeeper_data

# 客户端连接 ZooKeeper 的端口号（默认2181，需确保防火墙开放）
# 外部程序（如Kafka）通过此端口连接，例如 Kafka 配置 zookeeper.connect=IP:2181
clientPort=2181

# 限制单个客户端（如一个Kafka实例）的最大并发连接数（默认60）
# 若出现"连接数超限"报错，可取消注释并调大，例如：maxClientCnxns=100
#maxClientCnxns=60

# 以下是自动清理旧数据的配置（生产环境建议开启，避免磁盘占满）
# 请先阅读官方维护指南再开启：https://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance

# 自动保留的最新快照文件数量（取消注释后生效）
# 例如设置为3，表示只保留最近3个快照，更早的会被删除
#autopurge.snapRetainCount=3

# 自动清理任务的执行频率（单位：小时，取消注释后生效）
# 设置为0表示禁用自动清理；设置为1表示每1小时清理一次
#autopurge.purgeInterval=1

## 监控指标配置（对接Prometheus时使用，默认关闭）
#
# Prometheus监控 exporter 配置
#metricsProvider.className=org.apache.zookeeper.metrics.prometheus.PrometheusMetricsProvider
# 监控数据暴露的IP（0.0.0.0表示允许所有IP访问）
#metricsProvider.httpHost=0.0.0.0
# 监控数据暴露的端口（需开放此端口给Prometheus）
#metricsProvider.httpPort=7000
# 是否导出JVM信息（如内存、CPU使用情况）
#metricsProvider.exportJvmInfo=true

# 添加集群编号
server.0=192.168.201.112:2888:3888
server.1=192.168.201.113:2888:3888
```
指定集群编号192.168.201.112

```js
touch /kafka/zookeeper_data/myid
echo 0 > /kafka/zookeeper_data/myid
```
指定集群编号192.168.201.113

```js
touch /kafka/zookeeper_data/myid
echo 1 > /kafka/zookeeper_data/myid
```

然后这两台机器进入目录启动zookeeper
```js
cd /kafka/apache-zookeeper-3.9.4-bin/bin
./zkServer.sh start
```

这样就一个领导者一个工作者。（ leader领导者，follower：跟随者）
![image.png](/static/img/0d764541487333196e7f85cb571b9c2b.image.webp)

![image.png](/static/img/2f8e7c85cad8c6cf5a7884d4b2a8b882.image.webp)

# 三、修改Kakfa集群

创建Kafka的数据存储目录

```js
mkdir -p /kafka/data
```
修改配置文件
```js
vi /kafka/kafka_2.12-3.6.2/config/server.properties
```

服务器192.168.201.112配置
```js
############################# 服务器基本配置（ZooKeeper 模式） #############################
# 当前节点的唯一 ID（集群中不可重复）
broker.id=0

# ZooKeeper 集群连接地址（填你的 ZooKeeper 实际 IP:端口，单机就填一个）
zookeeper.connect=192.168.201.112:2181,192.168.201.113:2181
# 连接 ZooKeeper 的超时时间（默认 6000ms，无需修改）
zookeeper.connection.timeout.ms=6000


############################# socket 服务器设置（保留，无需修改） #############################
# 服务器监听的网络地址和端口（仅保留 PLAINTEXT 即可，删除 CONTROLLER 监听器）
listeners=PLAINTEXT://:9092
# broker 之间通信使用的监听器名称
inter.broker.listener.name=PLAINTEXT
# 向客户端暴露的连接地址（填 Kafka 实际 IP）
advertised.listeners=PLAINTEXT://192.168.201.112:9092
# 监听器与安全协议的映射（删除 CONTROLLER 相关映射）
listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL

# 网络和 IO 线程数（保留默认）
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600


############################# 日志存储基本配置（保留，无需修改） #############################
# 消息日志存储目录（你的配置是 /kafka/data，没问题）
log.dirs=/kafka/data
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.flush.interval.messages=10000
#log.flush.interval.ms=1000

log.retention.hours=168
#log.retention.bytes=1073741824
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
```

服务器192.168.201.113配置

```js
############################# 服务器基本配置（ZooKeeper 模式） #############################
# 当前节点的唯一 ID（集群中不可重复）
broker.id=1

# ZooKeeper 集群连接地址（填你的 ZooKeeper 实际 IP:端口，单机就填一个）
zookeeper.connect=192.168.201.112:2181,192.168.201.113:2181
# 连接 ZooKeeper 的超时时间（默认 6000ms，无需修改）
zookeeper.connection.timeout.ms=6000


############################# socket 服务器设置（保留，无需修改） #############################
# 服务器监听的网络地址和端口（仅保留 PLAINTEXT 即可，删除 CONTROLLER 监听器）
listeners=PLAINTEXT://:9092
# broker 之间通信使用的监听器名称
inter.broker.listener.name=PLAINTEXT
# 向客户端暴露的连接地址（填 Kafka 实际 IP）
advertised.listeners=PLAINTEXT://192.168.201.113:9092
# 监听器与安全协议的映射（删除 CONTROLLER 相关映射）
listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL

# 网络和 IO 线程数（保留默认）
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600


############################# 日志存储基本配置（保留，无需修改） #############################
# 消息日志存储目录（你的配置是 /kafka/data，没问题）
log.dirs=/kafka/data
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.flush.interval.messages=10000
#log.flush.interval.ms=1000

log.retention.hours=168
#log.retention.bytes=1073741824
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
```


然后进入bin目录启动kafka

```js
cd /kafka/kafka_2.12-3.6.2/bin/
```

```js
./kafka-server-start.sh -daemon ../config/server.properties
```

# 三、生产者和消费者测试
创建集群主题
```js
./kafka-topics.sh --bootstrap-server 192.168.201.112:9092,192.168.201.113:9092 --create --topic test_topic --partitions 3 --replication-factor 1
```
查看所有主题
```js
./kafka-topics.sh --bootstrap-server 192.168.201.112:9092,192.168.201.113:9092 --list
```

启动一个手动输入消息的生产者，向 test_topic 发送消息
```js
./kafka-console-producer.sh --bootstrap-server 192.168.201.112:9092,192.168.201.113:9092 --topic test_topic
```
启动消费者，订阅 test_topic 接收消息

```js
./kafka-console-consumer.sh --bootstrap-server 192.168.201.112:9092,192.168.201.113:9092 --topic test_topic --from-beginning
```

5.Kafka集群部署

# 一、使用kafka自带的命令创建主题
创建一个名为 test_topic 的主题，设置 3 个分区（提高并行度）、1 个副本（单机模式只能设 1，集群模式可设 2+）：

 创建主题命令：--bootstrap-server 指定 Kafka 地址，--topic 是主题名，--partitions 分区数，--replication-factor 副本数
```shell
./kafka-topics.sh --bootstrap-server 192.168.201.113:9092 --create --topic test_topic --partitions 3 --replication-factor 1
```

执行成功后会提示：Created topic test_topic.

![image.png](/static/img/08325bb87de48d727527e31bf5dfcf62.image.webp)

查看主题详情，确认创建成功：

```js
./kafka-topics.sh --bootstrap-server 192.168.201.113:9092 --describe --topic test_topic
```

![image.png](/static/img/cdaf844d2945fc1cfc0e00600c18049a.image.webp)

# 二、启动命令行生产者（发送消息）

启动一个手动输入消息的生产者，向 test_topic 发送消息（需要新开一个终端窗口执行，避免和后续消费者冲突）：
启动生产者命令：--bootstrap-server 指定 Kafka 地址，--topic 指定要发送的主题
```js
./kafka-console-producer.sh --bootstrap-server 192.168.201.113:9092 --topic test_topic
```
执行后会进入输入模式（显示 > 提示符），直接输入任意内容（比如 “hello kafka”“测试消息 123”），按回车就会发送消息；

![image.png](/static/img/f7dfa7aa0233d4a0773b7fd0a546ec94.image.webp)
想停止生产者，按 Ctrl+C 即可。

# 三、启动命令行消费者（接收消息）
再新开一个终端窗口，启动消费者，订阅 test_topic 接收消息（支持实时接收生产者发送的新消息，也能接收历史消息）：
启动消费者命令：--from-beginning 表示“从最早的消息开始接收”（不加则只收新消息）
```js
./kafka-console-consumer.sh --bootstrap-server 192.168.201.113:9092 --topic test_topic --from-beginning
```
执行后，消费者会自动接收 test_topic 里的所有消息（包括之前生产者发送的历史消息）；
此时回到生产者窗口，再输入新消息（比如 “这是新消息”），消费者窗口会实时显示这条消息，说明消息收发正常；


![image.png](/static/img/af812d3859f52a22a7ac0ad9b86c7205.image.webp)
想停止消费者，按 Ctrl+C 即可。

# 四、删除测试主题
如果后续不需要 test_topic，可以执行命令删除：

```js
./kafka-topics.sh --bootstrap-server 192.168.201.113:9092 --delete --topic test_topic
```

4.Kafka生产者和消费者测试

# 一、主题（Topic）管理命令
## 创建主题

基础创建（指定核心参数）

```js
./kafka-topics.sh --bootstrap-server 192.168.201.113:9092 \
  --create --topic test_topic \
  --partitions 3 \  # 分区数：根据并发处理需求设置，单机建议 1-5 个
  --replication-factor 1  # 副本数：单机只能设 1，集群建议 2-3 个（保证高可用）
```
带自定义配置创建（如消息保留时间、最大消息大小）

```js
./kafka-topics.sh --bootstrap-server 192.168.201.113:9092 \
  --create --topic long_retention_topic \
  --partitions 2 --replication-factor 1 \
  --config retention.ms=86400000 \  # 消息保留时长：24 小时（默认 7 天，单位：毫秒）
  --config max.message.bytes=1048576  # 单条消息最大大小：1MB（默认 1MB，单位：字节）
  --config cleanup.policy=delete  # 消息清理策略：删除（默认，可选 compact 压缩）
```


## 查看主题列表

查看所有主题
```js
./kafka-topics.sh --bootstrap-server 192.168.201.113:9092 --list
```

过滤查看指定前缀 / 关键词的主题
```js
# 示例：查看以“test_”开头的主题
./kafka-topics.sh --bootstrap-server 192.168.201.113:9092 --list | grep "test_"
```
## 查看主题详情

查看单个主题的完整信息（推荐）

输出包含：主题名称、分区数、副本分布（Leader/Follower 节点）、自定义配置、ISR 列表（同步副本集）等。
```js
./kafka-topics.sh --bootstrap-server 192.168.201.113:9092 \
  --describe --topic test_topic
```
查看所有主题的详情（不推荐，主题多时输出冗长）
```js
./kafka-topics.sh --bootstrap-server 192.168.201.113:9092 --describe
```
## 修改主题配置
新增 / 更新主题的自定义配置

```js
# 示例：将 test_topic 的消息保留时间延长至 7 天
./kafka-topics.sh --bootstrap-server 192.168.201.113:9092 \
  --alter --topic test_topic \
  --config retention.ms=604800000  # 7 天 = 7*24*3600*1000 毫秒
```
删除主题的自定义配置（恢复默认值）
```js
# 示例：删除“最大消息大小”的自定义配置，恢复默认值
./kafka-topics.sh --bootstrap-server 192.168.201.113:9092 \
  --alter --topic test_topic \
  --delete-config max.message.bytes
```
增加主题分区（仅支持扩容，不支持缩容）
```js
# 示例：将 test_topic 的分区数从 3 增加到 5
./kafka-topics.sh --bootstrap-server 192.168.201.113:9092 \
  --alter --topic test_topic \
  --partitions 5
```
分区扩容后，历史消息仅存在于原分区，新分区从扩容后开始接收消息，需确保业务能处理分区分布变化。


删除主题

```js
./kafka-topics.sh --bootstrap-server 192.168.201.113:9092 \
  --delete --topic test_topic
```
前提条件：
- Kafka 配置 delete.topic.enable 需设为 true（默认已开启）；
- 主题无正在运行的消费组，且无未提交的偏移量；
- 删除后不可恢复，需谨慎操作。


# 二、消费组（Consumer Group）管理命令
## 查看消费组列表
查看所有消费组
```js
./kafka-consumer-groups.sh --bootstrap-server 192.168.201.113:9092 --list
```
过滤查看指定消费组

```js
# 示例：查看包含“order”关键词的消费组
./kafka-consumer-groups.sh --bootstrap-server 192.168.201.113:9092 --list | grep "order"
```

## 查看消费组详情（偏移量、滞后消息数）
查看单个消费组的所有主题消费情况
```js
./kafka-consumer-groups.sh --bootstrap-server 192.168.201.113:9092 \
  --describe --group test_consumer_group
```
查看消费组在指定主题上的消费详情（精准查询）

```js
./kafka-consumer-groups.sh --bootstrap-server 192.168.201.113:9092 \
  --describe --group test_consumer_group \
  --topic test_topic
```

- 核心字段说明：
- 字段	               含义
- CURRENT-OFFSET	消费组当前已消费到的偏移量
- LOG-END-OFFSET	主题分区的最新消息偏移量（总消息数 = LOG-END-OFFSET - 起始偏移量）
- LAG	滞后消息数（未消费的消息数 = LOG-END-OFFSET - CURRENT-OFFSET），0 表示消费完成
- PARTITION	主题分区编号
- LEADER	负责处理该分区读写的 Broker 节点 ID


## 重置消费组偏移量（场景：重新消费、跳过错误消息）
重置前需确保：消费组无正在运行的消费者（或暂停消费者），避免偏移量被实时覆盖。

场景 1：重置到主题分区的最开始（重新消费所有历史消息）

```js
./kafka-consumer-groups.sh --bootstrap-server 192.168.201.113:9092 \
  --reset-offsets --group test_consumer_group \
  --topic test_topic \
  --to-earliest  # 核心参数：重置到最早偏移量
  --execute  # 必须加此参数才会实际执行（不加则仅预览重置结果）
```
场景 2：重置到指定时间点（如 1 小时前的消息）
```js
./kafka-consumer-groups.sh --bootstrap-server 192.168.201.113:9092 \
  --reset-offsets --group test_consumer_group \
  --topic test_topic \
  --to-datetime 2025-11-01T04:00:00.000  # 格式：YYYY-MM-DDTHH:MM:SS.fff
  --execute
```
场景 3：重置到指定偏移量（如将分区 0 的偏移量设为 100）

```js
./kafka-consumer-groups.sh --bootstrap-server 192.168.201.113:9092 \
  --reset-offsets --group test_consumer_group \
  --topic test_topic:0 \  # 指定主题的分区 0（格式：topic:partition）
  --to-offset 100  # 目标偏移量
  --execute
```
场景 4：重置到当前最新偏移量（跳过所有未消费消息）

```js
./kafka-consumer-groups.sh --bootstrap-server 192.168.201.113:9092 \
  --reset-offsets --group test_consumer_group \
  --topic test_topic \
  --to-latest  # 重置到最新偏移量
  --execute
```


## 删除消费组
前提条件：消费组为 “空闲状态”（无正在运行的消费者，且偏移量无关联业务），删除后不可恢复。
```js
./kafka-consumer-groups.sh --bootstrap-server 192.168.201.113:9092 \
  --delete --group test_consumer_group
```

# 三、其他命令
估算主题的消息总数
```js
./kafka-run-class.sh kafka.tools.GetOffsetShell \
  --bootstrap-server 192.168.201.113:9092 \
  --topic test_topic \
  --time -1  # -1 = 最新偏移量，-2 = 最早偏移量（总数 ≈ 各分区最新偏移量之和）
```

手动消费主题消息（调试用）

从历史消息开始消费（查看全量消息）

```js
./kafka-console-consumer.sh --bootstrap-server 192.168.201.113:9092 \
  --topic test_topic \
  --from-beginning \
  --group debug_consumer_group  # 临时消费组，避免干扰业务消费组
```

仅消费指定分区的消息

```js
./kafka-console-consumer.sh --bootstrap-server 192.168.201.113:9092 \
  --topic test_topic:0 \  # 仅消费分区 0
  --group debug_consumer_group
```
查看主题的配置列表（所有支持的配置项）

```js
./kafka-configs.sh --bootstrap-server 192.168.201.113:9092 \
  --describe --entity-type topics --entity-name test_topic
```

3.kafka命令

# 一、单机安装部署
先安装java JDK，高版本kafka只兼容javajdk11版本
```js
yum install -y java-11-openjdk
```
上传apache-zookeeper-3.9.4-bin.tar.gz和kafka_2.13-4.1.0.tgz包


进行解压

```js
tar -zxvf apache-zookeeper-3.9.4-bin.tar.gz
```
```js
tar -zxvf kafka_2.13-4.1.0.tgz
```

## 先修改zookeeper的配置
zoo.cfg是zookeeper的核心配置文件

把dataDir=/kafka/data修改，记得先创建好/kafka/data目录
```js
vi  /kafka/apache-zookeeper-3.9.4-bin/conf/zoo.cfg
```


```yml
# ZooKeeper 的时间基准单位（毫秒），所有时间相关配置都以它为倍数
# 比如后续的 initLimit、syncLimit 都是 "tickTime 的倍数"
tickTime=2000

# 从节点（Follower）启动后，与主节点（Leader）同步初始数据的最大超时时间
# 计算方式：initLimit × tickTime = 10 × 2000ms = 20秒
# 超过这个时间未同步成功，从节点会被判定为连接失败
initLimit=10

# 运行中，从节点与主节点之间通信的最大超时时间（如请求响应超时）
# 计算方式：syncLimit × tickTime = 5 × 2000ms = 10秒
# 超过这个时间未收到响应，主节点会认为从节点失联
syncLimit=5

# 数据存储目录（核心！必须改）
# 存储内容：内存快照（snapshot）、事务日志（记录所有写操作，用于恢复数据）
# 注意：/tmp 是临时目录，Linux 重启后会自动清空，生产环境必须改成永久目录
# 示例：dataDir=/data/zookeeper（需提前创建该目录并授权）
dataDir=/kafka/zookeeper_data

# 客户端连接 ZooKeeper 的端口号（默认2181，需确保防火墙开放）
# 外部程序（如Kafka）通过此端口连接，例如 Kafka 配置 zookeeper.connect=IP:2181
clientPort=2181

# 限制单个客户端（如一个Kafka实例）的最大并发连接数（默认60）
# 若出现"连接数超限"报错，可取消注释并调大，例如：maxClientCnxns=100
#maxClientCnxns=60

# 以下是自动清理旧数据的配置（生产环境建议开启，避免磁盘占满）
# 请先阅读官方维护指南再开启：https://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance

# 自动保留的最新快照文件数量（取消注释后生效）
# 例如设置为3，表示只保留最近3个快照，更早的会被删除
#autopurge.snapRetainCount=3

# 自动清理任务的执行频率（单位：小时，取消注释后生效）
# 设置为0表示禁用自动清理；设置为1表示每1小时清理一次
#autopurge.purgeInterval=1

## 监控指标配置（对接Prometheus时使用，默认关闭）
#
# Prometheus监控 exporter 配置
#metricsProvider.className=org.apache.zookeeper.metrics.prometheus.PrometheusMetricsProvider
# 监控数据暴露的IP（0.0.0.0表示允许所有IP访问）
#metricsProvider.httpHost=0.0.0.0
# 监控数据暴露的端口（需开放此端口给Prometheus）
#metricsProvider.httpPort=7000
# 是否导出JVM信息（如内存、CPU使用情况）
#metricsProvider.exportJvmInfo=true
```


进入bin目录启动zookeeper
```js
cd /kafka/apache-zookeeper-3.9.4-bin/bin
```
启动
```js
./zkServer.sh status
```

![image.png](/static/img/0c8fb61dd3e5c46290ebdd36e6202919.image.webp)

测试链接
```js
./zkCli.sh
```

## Kafka配置文件

![image.png](/static/img/90beb0035a29e97a4a947635e86d20a9.image.webp)
```js
# kafka-server-start.sh
Kafka Broker 启动程序，需指定配置文件（如 server.properties）才能启动服务端节点。

# kafka-server-stop.sh
Kafka Broker 停止程序，用于优雅关闭运行中的 Broker 进程，避免数据丢失。

# kafka-topics.sh
Topic 全生命周期管理程序，可创建、删除、修改 Topic（如调整分区数）、查看 Topic 列表/详情。

# kafka-console-producer.sh
命令行模拟生产者程序，手动输入消息并发送到指定 Topic，常用于测试消息生产流程。

# kafka-console-consumer.sh
命令行模拟消费者程序，订阅指定 Topic 并消费消息，支持从最新位置或最早位置开始消费，用于测试消息消费流程。
```


```js
/kafka/kafka_2.13-4.1.0/config
```

最核心的配置文件：server.properties（Broker 必用）、producer.properties（生产者）、consumer.properties（消费者）。



```js
vi server.properties
```
使用ZooKeeper
```js
############################# 服务器基本配置（ZooKeeper 模式） #############################
# 当前节点的唯一 ID（集群中不可重复）
broker.id=1

# ZooKeeper 集群连接地址（填你的 ZooKeeper 实际 IP:端口，单机就填一个）
zookeeper.connect=192.168.201.113:2181
# 连接 ZooKeeper 的超时时间（默认 6000ms，无需修改）
zookeeper.connection.timeout.ms=6000


############################# socket 服务器设置（保留，无需修改） #############################
# 服务器监听的网络地址和端口（仅保留 PLAINTEXT 即可，删除 CONTROLLER 监听器）
listeners=PLAINTEXT://:9092
# broker 之间通信使用的监听器名称
inter.broker.listener.name=PLAINTEXT
# 向客户端暴露的连接地址（填 Kafka 实际 IP）
advertised.listeners=PLAINTEXT://192.168.201.113:9092
# 监听器与安全协议的映射（删除 CONTROLLER 相关映射）
listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL

# 网络和 IO 线程数（保留默认）
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600


############################# 日志存储基本配置（保留，无需修改） #############################
# 消息日志存储目录（你的配置是 /kafka/data，没问题）
log.dirs=/kafka/data
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.flush.interval.messages=10000
#log.flush.interval.ms=1000

log.retention.hours=168
#log.retention.bytes=1073741824
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
```

不使用ZooKeeper
```yml
############################# 服务器基本配置 #############################

# 当前服务器的角色（KRaft 模式必填）
# 可选值：broker（仅作为消息代理）、controller（仅作为控制器）、broker,controller（同时作为代理和控制器，单机模式常用）
process.roles=broker,controller

# 当前节点的唯一 ID（集群中所有节点必须不同，如 1、2、3...）
node.id=1

# 控制器集群的连接地址（KRaft 模式必填，用于节点间通信）
# 格式：主机:端口（若为集群，用逗号分隔多个控制器）
controller.quorum.bootstrap.servers=192.168.201.113:9093


#############################  socket 服务器设置 #############################

# 服务器监听的网络地址和端口（KRaft 模式需至少包含控制器的监听器）
# 格式：监听器名称://主机:端口（多个监听器用逗号分隔）
# 示例中：PLAINTEXT 用于客户端和 broker 通信，CONTROLLER 用于控制器内部通信
listeners=PLAINTEXT://:9092,CONTROLLER://:9093

# broker 之间通信使用的监听器名称（需与上面的监听器名称对应）
inter.broker.listener.name=PLAINTEXT

# 向客户端暴露的连接地址（若不设置，默认使用 listeners 的值）
# 生产环境需填写实际可访问的 IP:端口（如 192.168.1.100:9092），否则客户端可能连接失败
advertised.listeners=PLAINTEXT://192.168.201.113:9092,CONTROLLER://192.168.201.113:9093

# 控制器使用的监听器名称（KRaft 模式必填，需与 listeners 中的控制器监听器对应）
controller.listener.names=CONTROLLER

# 监听器名称与安全协议的映射（默认无需修改，PLAINTEXT 表示无加密）
# 格式：监听器名称:协议（如 CONTROLLER 用 PLAINTEXT 协议）
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL

# 用于接收网络请求和发送响应的线程数（根据服务器 CPU 调整，默认 3 即可）
num.network.threads=3

# 用于处理请求（可能包含磁盘 IO）的线程数（建议大于等于磁盘数量，默认 8）
num.io.threads=8

# socket 发送缓冲区大小（默认 100KB，无需修改）
socket.send.buffer.bytes=102400

# socket 接收缓冲区大小（默认 100KB，无需修改）
socket.receive.buffer.bytes=102400

# 服务器可接受的最大请求大小（防止内存溢出，默认 100MB）
socket.request.max.bytes=104857600


############################# 日志存储基本配置 #############################

# 日志文件（消息数据）的存储目录（多个目录用逗号分隔）
# 注意：/tmp 是临时目录，重启后数据会丢失，生产环境必须改为永久目录（如 /data/kafka/logs）
log.dirs=/kafka/data

# 新建主题时的默认分区数（分区越多，并行消费能力越强，但文件数量也越多）
num.partitions=1

# 每个数据目录用于日志恢复（启动时）和刷新（关闭时）的线程数
# 若目录位于 RAID 阵列，建议增大此值
num.recovery.threads.per.data.dir=1


############################# 内部主题设置 #############################
# 内部主题（如消费者偏移量、事务状态）的副本数
# 开发测试可用 1，生产环境建议大于 1（如 3），确保高可用
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  # 事务状态日志的最小同步副本数


############################# 日志刷新策略 #############################

# 消息会立即写入文件系统，但默认延迟刷新到磁盘（由操作系统缓存管理）
# 以下配置控制数据刷新到磁盘的策略，需权衡：
# 1. 持久性：若不启用副本，未刷新的数据可能丢失
# 2. 延迟：刷新间隔过大会导致刷新时延迟飙升
# 3. 吞吐量：刷新是昂贵操作，间隔过小会导致频繁磁盘寻道

# 累计多少条消息后强制刷新到磁盘（默认注释，即不按消息数刷新）
#log.flush.interval.messages=10000

# 消息在内存中最多存放多久后强制刷新到磁盘（默认注释，即不按时间刷新）
#log.flush.interval.ms=1000


############################# 日志保留策略 #############################

# 以下配置控制日志段的清理策略：按时间或大小清理
# 只要满足任一条件，日志段就会被删除（从日志末尾开始）

# 日志文件可被删除的最小存活时间（默认 168 小时，即 7 天）
log.retention.hours=168

# 基于大小的日志保留策略（默认注释，即不按大小清理）
# 若启用，当剩余日志大小低于此值时，停止清理（单位：字节，1GB=1073741824）
#log.retention.bytes=1073741824

# 单个日志段文件的最大大小（达到此值时创建新的日志段，默认 1GB）
log.segment.bytes=1073741824

# 检查日志段是否符合清理条件的时间间隔（默认 300000ms，即 5 分钟）
log.retention.check.interval.ms=300000
```

1. process.roles：KRaft 模式必填，单机测试用 broker,controller，集群需区分角色（部分节点为 controller，部分为 broker）。
2. node.id：集群中每个节点的唯一 ID，不能重复。
3. controller.quorum.bootstrap.servers：集群模式需填写所有控制器节点的地址（如 node1:9093,node2:9093）。
4. log.dirs：必须修改为永久目录（如 /data/kafka/logs），否则重启后数据丢失。
5. advertised.listeners：生产环境需填写服务器的实际 IP（如 PLAINTEXT://192.168.1.100:9092），否则外部客户端无法连接。

启动 Kafka
```js
cd /kafka/kafka_2.13-4.1.0/bin  # 你的 Kafka bin 目录路径
```

后台启动，日志会输出到 logs 目录，不占用终端
```js
./kafka-server-start.sh -daemon ../config/server.properties
```

```js
ps -ef | grep kafka
# 若输出包含“kafka.Kafka”，说明进程正常
```

2.Kafka安装部署

# 一、消息队列的相关几个概念

- 消息（Message):网络中的两台计算机或者两个通讯设备之间传递的数据就是消息，如：文本、音频、视频等。
- 队列（Queue)：一种特殊的线性表（理解：有一定顺序的元素排列，如排队）特殊之处在于它只允许在首部删除元素（队首取出数据），在尾部添加元素（队尾添加数据）。类似羽毛球筒，在一头放，另一头取
- 消息队列（MQ）：保存消息的队列，是消息在传输过程中的容器。主要提供生产和消费接口供外部调用，进行数据的存储和获取。

# 二、消息队列(MQ)分类
- 主要分为两类：点对点（PeertoPeer)、发布/订阅(Publish/Subscribe）
- 共同点：消息生产者（Producer)生产消息发送到队列中，然后消息消费者（Consumer）从队列中读取并消费消息。
- 不同点：


- 1)点对点：
- 组成：消息队列(Queue)、发送者(Sender)、接收者(Receiver】
- 一个生产者生产的消息只能有一个消费者，消息一旦被消费，消息就不在消息队列中了，如：打电话，
- 即发送到消息队列的消息能且只能被一个接收者接收
- 2).发布/订阅：
- 组成：消息队列(Queue）、发布者（Publisher)、订阅者（Subscriber)、主题（Topic）需要有讨论的主题，如这堂课讲的什么主题，这份报纸讲的什么新闻主题
- 每个消息可以有多个消费者，彼此互不影响，即发布到消息队列的消息能被多个接收者（订阅者）接收。
- 如：我发布一个微博，关注我的人都能看到，很多人订阅了同一个公众号，公众号内容一更新，很多人都能收到

# 三、常见消息系统
- ActiveMQ：历史悠久，支持性较好，性能相对不高
- RabbitMQ：可靠性高、安全
- Kafka：分布式、高性能、高吞吐量、跨语言
- RocketMQ：阿里开源的消息中间件，纯Java实现







![image.png](/static/img/f35f1407b3025ad5abd8f3e9df430bc1.image.webp)




![image.png](/static/img/0dd627a179617893d56592ef2e68045d.image.webp)