Kafka配置

## CentOS7 - Kafka
##### 三台CentOS7.9配置

1. 下载kafka的压缩包
2. 解压压缩包到指定目录
3. 配置kafka的bin目录到环境变量

``` /etc/profile
export KAFKA_HOME=kafka安装路径
export PATH=$PATH:$KAFKA_HOME/bin
```
重载配置
`source /etc/profile`

4. 配置kafka的配置文件（编辑 server.properties文件）
   
```编辑kafka配置文件
sudo vim server.properties
```
* #### broker.id(必须修改):

* 这是 Kafka Broker 的唯一标识符。集群中每个 Broker 必须不同。
  * 对于单节点测试，可以设置为 0。
  * 如果是集群，需要为每个节点分配一个唯一的整数 ID（如 0, 1, 2...）
* #### listeners(必须修改):
  * 指定 Broker 监听客户端连接的地址和协议。
  * 将 PLAINTEXT://:9092修改为你的服务器 IP 或主机名（如果想让外部访问），或者保持 localhost仅限本地访问。
  * 重要：如果 Kafka 和 Zookeeper 在同一台机器且仅用于本地测试，localhost通常足够。如果 Kafka Broker 需要被集群内其他节点或外部应用访问，必须使用服务器的真实 IP 或可解析的主机名，并且确保防火墙开放相应端口（默认 9092）。
 * #### zookeeper.connect(必须修改):
   * **这是连接 Kafka 和 Zookeeper 的最关键配置！**
   * 指向你正在运行的 Zookeeper 实例的连接字符串。
   * 格式：host:port[,host:port,...]。如果是 Zookeeper 集群，用逗号分隔多个节点。
   * 确认你的 Zookeeper 监听地址和端口（默认 **2181**）。
 * ##### log.dirs(建议修改):
   * 指定 Kafka 存储消息数据（日志段）的目录。强烈建议指向一个独立的、高性能、大容量的磁盘分区。
   * 默认是 /tmp/kafka-logs，这不是一个好选择，因为 /tmp可能在系统重启后被清空！
   * 创建一个专门的目录，例如 /var/lib/kafka/data：
 * ##### advertised.listeners(建议修改，尤其在生产或网络环境):
   * 告诉 Producer 和 Consumer 实际连接到哪个地址。
   * 通常设置为和 listeners相同，或者设置为客户端能够访问的地址（如果服务器有多个网卡或经过 NAT）。
   * 对于简单测试，可以暂时不设置或设置成和 listeners一样。
```server.properties
broker.id=0
listeners=PLAINTEXT://你的服务器IP:9092
zookeeper.connect=master:2181,slave1:2181,slave2:2181
log.dirs=kafka路径/logs
```

```禁用不依赖Zookeeper启动模式（KRaft）
# 禁用 KRaft
# process.roles=broker,controller
# node.id=1
# controller.quorum.voters=1@localhost:9093
# controller.listener.names=CONTROLLER
```

5. scp到每一台机器
6. 在每一台机器启动kafka-broker（确保Zookeeper已经启动在后台）

```后台启动kafka-broker-推荐
#默认不占用前台，后台运行
nohup bin/kafka-server-start.sh config/server.properties &
#可以回到前台查看实时日志输出
tail -f nohup.out
```

```前台启动kafka-broker
# 在 master 节点
kafka-server-start.sh /opt/module/kafka/config/server.properties
# 在 slave1 节点
kafka-server-start.sh /opt/module/kafka/config/server.properties
# 在 slave2 节点
kafka-server-start.sh /opt/module/kafka/config/server.properties
```

>i **确保**
>
> 观察启动日志，看是否有 INFO [KafkaServer id=0] started (kafka.server.KafkaServer)之类的成功信息。
>按 Ctrl+C停止 Broker。

> jps指令 确保kafka的进程已经启动

7. 创建Topic

```指令
# 查看 broker 列表（需要所有节点都启动）
bin/kafka-broker-api-versions.sh --bootstrap-server 10.155.151.141:9092 | grep "10.155.151"

# 创建测试 Topic
bin/kafka-topics.sh --create \
  --topic test-topic \
  --bootstrap-server 10.155.151.141:9092 \
  --partitions 3 \
  --replication-factor 3

# 查看 Topic 详情（重点看 Replica 分布）
bin/kafka-topics.sh --describe --topic test-topic --bootstrap-server 10.155.151.141:9092
```
##### `--topic` 设置Topic的名称、`--partitions` 设置分区数、`--replication-factor` 设置副本数