JDK
需提前安装 Java 并配置 JAVA_HOME 环境变量，确保需要的 Java 环境正常。
D:\\Users\\Administrator>java –version
java 11.0.6 2020-01-14 LTS
Java(TM) SE Runtime Environment 18.9 (build 11.0.6+8-LTS)
Java HotSpot(TM) 64-Bit Server VM 18.9 (build 11.0.6+8-LTS, mixed mode)
D:\\Users\\Administrator>javac –version
javac 11.0.6
下载
访问 Apache Kafka 官网，选择二进制文件 Binary Downloads，下载 kafka_2.12-3.6.1.tgz（支持 Windows/Linux）。这里根据自己的需要下载对应的版本，并且要注意需要和你的 Java 的版本要兼容。
安装
解压到指定目录即可。比如我解压到 D:\\kafka_2.12-3.6.1
Administrator@DESKTOP-UUSNMAC MINGW64 /d/kafka_2.12-3.6.1
-rw-r–r– 1 Administrator 197121 15030 Nov 24 2023 LICENSE
-rw-r–r– 1 Administrator 197121 28184 Nov 24 2023 NOTICE
drwxr-xr-x 1 Administrator 197121 0 Sep 8 15:42 bin/
drwxr-xr-x 1 Administrator 197121 0 Sep 8 15:42 config/
drwxr-xr-x 1 Administrator 197121 0 Sep 8 15:42 libs/
drwxr-xr-x 1 Administrator 197121 0 Sep 8 15:42 licenses/
drwxr-xr-x 1 Administrator 197121 0 Sep 8 15:42 site-docs/
配置
创建日志目录
修改配置文件
编辑 config\\server.properties，更新目录位置参数（Windows 环境下路径使用双反斜杠分隔，避免无法识别）：
# 日志存储路径（使用双反斜杠）
log.dirs=D:\\\\kafka_2.12-3.6.1\\\\kafka-logs
启动
启动Zookeeper
cd D:\\kafka_2.12-3.6.1>
bin\\windows\\zookeeper-server-start.bat config\\zookeeper.properties
启动Kafka
cd D:\\kafka_2.12-3.6.1>
bin\\windows\\kafka-server-start.bat config\\server.properties
成功标志：分别有对应的日志输出。
验证
CMD验证
可以直接在 cmd 命令行窗口验证
producer
D:\\kafka_2.12-3.6.1\\bin\\windows>kafka-console-producer.bat –bootstrap-server localhost:9092 –topic test-topic
>Hello kafka!
consumer
D:\\kafka_2.12-3.6.1\\bin\\windows>kafka-console-consumer.bat –bootstrap-server localhost:9092 –topic test-topic –from-beginning
Hello kafka!
Python脚本验证
也可以安装下面的依赖包以后
pip install kafka-python
使用 Python 写个简单示例进行验证。
producer.py
from kafka import KafkaProducer
producer = KafkaProducer(
bootstrap_servers='localhost:9092',
value_serializer=lambda v: str(v).encode('utf-8')
)
# 发送消息，key 和 value 必须是字节类型
future = producer.send('my_topic', value='hello kafka!')
# 等待发送结果，确认是否成功
result = future.get(timeout=10)
print(f"发送成功: topic={result.topic}, partition={result.partition}, offset={result.offset}")
producer.close()
发送成功: topic=my_topic, partition=0, offset=7
consumer.py
from kafka import KafkaConsumer
consumer = KafkaConsumer(
'my_topic',
bootstrap_servers='localhost:9092',
group_id='my_group',
auto_offset_reset='earliest', # 从最早的消息开始读
value_deserializer=lambda m: m.decode('utf-8')
)
for message in consumer:
print(f"收到消息: {message.value}")
收到消息: hello kafka!
