目录
一. Kafka安装包获取
1. 官网地址
2. 百度网盘链接
二. 环境要求
1. Java 运行环境
(1) 对 java 环境变量进行配置
(2) 下载完毕之后进行解压
三. 启动Zookeeper
四. 启动Kafka
(1) 修改Conf下的server.properties文件，修改kafka的日志文件路径
(2) Kafka开启远程连接
五. kafka-windwos 自启动脚本
六. python-demo生产者消费者案例
（1）生产者
（2）消费者
七. Docker搭建kafka-map (数据可视化)
八. 备注
Kafka是一款流行分布式消息分布订阅系统，除Kafka之外还有MQ、Redis等。我们可以把消息队列视为一个管道，管道的两端分别是消息生产者(producer)和消息消费者(consumer)，消息生产者产生日志等消息后可以发送到管道中，这时消息队列可以驻留在内存或者磁盘上，直到消费者来把它读走为止。
上述就是Kafka的一个概括，我们只需要了解一下Kafka的架构和一些专业术语即可，下面就来介绍一下Kafka 中一些专业术语。
Producer：消息生产者，负责把产生的消息发送到Kafka服务器上。
Consumer：消息消费者，从Kafka服务器读取消息。
Consumer Group：消费者群组，每个消息消费者可以划分为一个特定的群组。
Topic：这是Kafka使用中非常重要的一个术语，它相当于消息的"身份标识"，消息生产者产生消息时会给它贴上一个Topic标签，当消息消费者需要读取消息时，可以根据这个Topic读取特定的数据。
Broker：Kafka集群中包含的服务器。
一. Kafka安装包获取
1. 官网地址
Apache Kafka
https://kafka.apache.org/downloads
下载版本为kafka_2.13-3.5.0
2. 百度网盘链接
百度网盘 请输入提取码 (baidu.com)https://pan.baidu.com/share/init?surl=qD06L8_OLbe7NFmQI3Ja6g&pwd=2024
二. 环境要求
在安装 Kafka 之前，需要满足以下环境要求：
1. Java 运行环境
Kafka 是使用 Java 语言编写的，因此需要在安装 Kafka 之前先安装 Java 运行环境。Kafka 支持 Java 8 及以上版本。可以通过以下命令检查 Java 运行环境的版本：
java -version(1) 对 java 环境变量进行配置
(2) 下载完毕之后进行解压
因为Kafka的运行依赖于 Zookeeper，所以还需要下并安装Zookeeper，ZooKeeper和Kafka版本之间有一定的对应关系，不同版本的ZooKeeper和Kafka可以相互兼容，但需要满足一定的条件。
Kafka 2.2.0 开始支持使用内置的ZooKeeper替代外部ZooKeeper。 所以3.5.0是不需要安装Zookeeper的,直接解压即可。
三. 启动Zookeeper
因为Kafka中的Broker注册，Topic注册，以及负载均衡都是在Zookeeper中管理，所以需要先启动内置的Zookeeper
打开Conf文件下的zookeeper.properties文件，修改dataDir目录路径
dataDir=D:\Kafka\kafka_2.13-3.5.0\zookeeperData
执行启动Zookeeper命令
.\bin\windows\zookeeper-server-start.bat .\config\zookeeper.properties当看到绑定到IP地址为0.0.0.0、端口号为2181的地址，表示ZooKeeper服务器监听在该地址，启动成功
四. 启动Kafka
(1) 修改Conf下的server.properties文件，修改kafka的日志文件路径
(2) Kafka开启远程连接
新开一个命令行窗口，在之前的目录中输入启动命令
.\bin\windows\kafka-server-start.bat .\config\server.properties 
五. kafka-windwos 自启动脚本
@echo off
REM 设置绝对路径
set KAFKA_HOME=D:\Kafka\kafka_2.13-3.5.0
set ZOOKEEPER_CONFIG=%KAFKA_HOME%\config\zookeeper.properties
set KAFKA_CONFIG=%KAFKA_HOME%\config\server.properties
REM 删除 kafka_log 目录及其内容
if exist %KAFKA_HOME%\kafka_log (
echo 删除 kafka_log 目录...
rmdir /s /q %KAFKA_HOME%\kafka_log
)
REM 删除 zookeeperData 目录
if exist %KAFKA_HOME%\bin\windows\Kafkakafka_2.13-3.5.0zookeeperData (
echo 删除 Kafkakafka_2.13-3.5.0zookeeperData 目录...
rmdir /s /q %KAFKA_HOME%\bin\windows\Kafkakafka_2.13-3.5.0zookeeperData
)
REM 启动Zookeeper
cd /d %KAFKA_HOME%\bin\windows
start /b zookeeper-server-start.bat %ZOOKEEPER_CONFIG%
REM 等待Zookeeper启动
timeout /t 10 /nobreak
REM 检查 meta.properties 文件
if exist %KAFKA_HOME%\kafka_log\meta.properties (
echo "Meta properties already exist. Starting Kafka server..."
) else (
echo "Meta properties not found. Starting Kafka server for the first time..."
)
REM 启动Kafka服务器
start /b kafka-server-start.bat %KAFKA_CONFIG%
exit
将脚本后缀改为bat，放到windwos自启中
六. python-demo生产者消费者案例
（运行如下代码可以对kafka数据进行推送与拉取）
（1）生产者
from kafka import KafkaProducer
import json
producer = KafkaProducer(bootstrap_servers='192.168.14.93:9092',
value_serializer=lambda v: json.dumps(v).encode('utf-8'))
# kafka 队列生产者
def producer_demo():
# 配置Kafka生产者
producer = KafkaProducer(bootstrap_servers='192.168.14.93:9092',
value_serializer=lambda v: json.dumps(v).encode('utf-8'))
# 发送消息到指定的topic
producer.send('my-topic', {'data_time':'2024-09-10 17:06:12',
'data_image':'http://192.168.7.51:9001/var/model_algorithm_package/output/17/2024-09-10-17-06-11.jpg',
'data_result':
"['[323, 356, 437, 665, 0, [226, 255, 0], 0.5512826]', '[1514, 238, 1619, 447, 0, [226, 255, 0], 0.5094569]']"
})
# 关闭生产者连接
producer.close()
if __name__ == '__main__':
while 1:
producer_demo()
（2）消费者
import json
from kafka import KafkaConsumer
lisr = []
# kafka 队列消费者
def consumer_demo():
# 配置Kafka消费者
consumer = KafkaConsumer('my-topic',
bootstrap_servers='192.168.14.93:9092',
# auto_offset_reset='latest', # 从最新消息开始消费
auto_offset_reset='earliest', # 从最新消息开始消费
enable_auto_commit=True, # 自动提交offset
group_id='my-consumer-group') # 指定消费者组
# 消费消息
for message in consumer:
res = message.value.decode('utf-8')
try:
# 解析 JSON 字符串为 Python 字典
parsed_dict = json.loads(res)
# 打印解析后的字典
print(parsed_dict)
except Exception as e:
print(f"Error: {e}, 消息内容: {res}")
if __name__ == '__main__':
consumer_demo()
七. Docker搭建kafka-map (数据可视化)
docker run -d --name kafka-map \
-p 9001:8080 \
-v /opt/kafka-map/data:/usr/local/kafka-map/data \
-e DEFAULT_USERNAME=admin \
-e DEFAULT_PASSWORD=admin \
--restart always dushixiang/kafka-map:latest
登录端口地址为9001
可以对相应分区数据进行拉取
八. 备注
Kafka定位为分布式消息发布-订阅系统，提及分布式就可以想象，只有当在多节点环境下才能最大的发挥它的价值，前面所介绍的Kafka配置方式都是基于单节点的配置，由于本文主要是为了系统的梳理一下Kafka的配置及使用，因此对于配置这一块不再花费大的篇幅去详细介绍，如果需要到多节点配置Kafka可以自行查阅其他资料。
