在CentOS上配置Kafka主题涉及多个步骤，包括安装必要的软件、配置Kafka服务器、启动服务以及创建和配置Kafka主题。以下是详细的步骤：

1. 安装Java

Kafka需要Java环境，首先需要安装Java。

sudo yum install java-1.8.0-openjdk-devel

验证Java安装：

java -version

2. 下载和解压Kafka

从Apache Kafka官方网站下载最新版本的Kafka，并解压。

wget https://downloads.apache.org/kafka/3.4.0/kafka_2.13-3.4.0.tgz
tar -xzf kafka_2.13-3.4.0.tgz
cd kafka_2.13-3.4.0

3. 配置Zookeeper

Kafka使用Zookeeper进行集群管理和元数据存储。

3.1 安装Zookeeper

wget https://downloads.apache.org/zookeeper/zookeeper-3.8.0/apache-zookeeper-3.8.0-bin.tar.gz
tar -xzf apache-zookeeper-3.8.0-bin.tar.gz
cd apache-zookeeper-3.8.0

3.2 配置Zookeeper

在每个Zookeeper节点上编辑zoo.cfg文件。

cp zoo_sample.cfg zoo.cfg

修改配置：

dataDir=/var/lib/zookeeper
server.1=kafka01:2888:3888
server.2=kafka02:2888:3888
server.3=kafka03:2888:3888

创建myid文件：

echo 1 > /var/lib/zookeeper/myid

3.3 启动Zookeeper

bin/zkServer.sh start

验证启动状态：

bin/zkServer.sh status

4. 配置Kafka

Kafka的主要配置文件是server.properties，位于config目录下。

4.1 配置broker.id

每个broker需要一个唯一的ID。

broker.id=0

4.2 配置listeners

指定Kafka broker监听的地址和端口。

listeners=PLAINTEXT://your_server_ip:9092
advertised.listeners=PLAINTEXT://your_server_ip:9092

将your_server_ip替换为你的服务器IP地址。

4.3 配置log.dirs

指定Kafka日志文件的存储目录。

log.dirs=/tmp/kafka-logs

4.4 配置zookeeper.connect

指定Zookeeper的连接字符串。

zookeeper.connect=localhost:2181

如果你有多个Zookeeper节点，可以这样配置：

zookeeper.connect=zookeeper1:2181,zookeeper2:2181,zookeeper3:2181

5. 启动Kafka Broker

在另一个终端窗口中启动Kafka Broker。

bin/kafka-server-start.sh config/server.properties

6. 创建Topic

创建一个Topic来测试Kafka。

bin/kafka-topics.sh --create --topic test --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1

7. 验证Kafka

7.1 发送消息

使用以下命令发送消息到test主题：

bin/kafka-console-producer.sh --topic test --bootstrap-server localhost:9092

在提示符下输入消息并按回车键发送。

7.2 接收消息

使用以下命令从test主题消费消息：

bin/kafka-console-consumer.sh --topic test --from-beginning --bootstrap-server localhost:9092

你应该能够看到发送的消息。

8. 停止Kafka和Zookeeper

停止Kafka Broker：

bin/kafka-server-stop.sh

停止Zookeeper：

bin/zookeeper-server-stop.sh

通过以上步骤，你应该能够在CentOS上成功配置和运行Kafka。