一、使用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.

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

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

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

启动一个手动输入消息的生产者，向 test_topic 发送消息（需要新开一个终端窗口执行，避免和后续消费者冲突）： 启动生产者命令：--bootstrap-server 指定 Kafka 地址，--topic 指定要发送的主题

js
./kafka-console-producer.sh --bootstrap-server 192.168.201.113:9092 --topic test_topic

执行后会进入输入模式（显示 > 提示符），直接输入任意内容（比如 “hello kafka”“测试消息 123”），按回车就会发送消息；

想停止生产者，按 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 里的所有消息（包括之前生产者发送的历史消息）； 此时回到生产者窗口，再输入新消息（比如 “这是新消息”），消费者窗口会实时显示这条消息，说明消息收发正常；

想停止消费者，按 Ctrl+C 即可。

四、删除测试主题

如果后续不需要 test_topic，可以执行命令删除：

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