当前位置：首页 > news >正文

Kafka-Windows搭建全流程（环境，安装包，编译，消费案例，远程连接，服务自启，可视化工具）

news 2024/10/23 11:06:10

一. 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 Kafkahttps://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 offREM 设置绝对路径
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.propertiesREM 删除 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 /nobreakREM 检查 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 jsonproducer = 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'))# 发送消息到指定的topicproducer.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 KafkaConsumerlisr = []# 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,      # 自动提交offsetgroup_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可以自行查阅其他资料。

查看全文

http://www.mrgr.cn/news/56694.html