做网站前期需要准备什么,wp网站开发,用jsp做网站步骤,住房和城乡建设厅施工员证目录
一. Kafka安装包获取 1. 官网地址 2. 百度网盘链接
二. 环境要求
1. Java 运行环境
(1) 对 java 环境变量进行配置
(2) 下载完毕之后进行解压
三. 启动Zookeeper
四. 启动Kafka
(1) 修改Conf下的server.properties文件#xff0c;修改kafka的日志文件路径
(2)…
目录
一. 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读取特定的数据。
BrokerKafka集群中包含的服务器。 一. Kafka安装包获取 1. 官网地址
Apache Kafkahttps://kafka.apache.org/downloads 下载版本为kafka_2.13-3.5.0 2. 百度网盘链接
百度网盘 请输入提取码 (baidu.com)https://pan.baidu.com/share/init?surlqD06L8_OLbe7NFmQI3Ja6gpwd2024
二. 环境要求 在安装 Kafka 之前需要满足以下环境要求
1. Java 运行环境
Kafka 是使用 Java 语言编写的因此需要在安装 Kafka 之前先安装 Java 运行环境。Kafka 支持 Java 8 及以上版本。可以通过以下命令检查 Java 运行环境的版本
java -version
(1) 对 java 环境变量进行配置 (2) 下载完毕之后进行解压 因为Kafka的运行依赖于 Zookeeper所以还需要下并安装ZookeeperZooKeeper和Kafka版本之间有一定的对应关系不同版本的ZooKeeper和Kafka可以相互兼容但需要满足一定的条件。 Kafka 2.2.0 开始支持使用内置的ZooKeeper替代外部ZooKeeper。 所以3.5.0是不需要安装Zookeeper的,直接解压即可。 三. 启动Zookeeper 因为Kafka中的Broker注册Topic注册以及负载均衡都是在Zookeeper中管理所以需要先启动内置的Zookeeper 打开Conf文件下的zookeeper.properties文件修改dataDir目录路径
dataDirD:\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_HOMED:\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_servers192.168.14.93:9092,value_serializerlambda v: json.dumps(v).encode(utf-8))
# kafka 队列生产者
def producer_demo():# 配置Kafka生产者producer KafkaProducer(bootstrap_servers192.168.14.93:9092,value_serializerlambda 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_servers192.168.14.93:9092,# auto_offset_resetlatest, # 从最新消息开始消费auto_offset_resetearliest, # 从最新消息开始消费enable_auto_commitTrue, # 自动提交offsetgroup_idmy-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(fError: {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_USERNAMEadmin \-e DEFAULT_PASSWORDadmin \--restart always dushixiang/kafka-map:latest登录端口地址为9001 可以对相应分区数据进行拉取 八. 备注
Kafka定位为分布式消息发布-订阅系统提及分布式就可以想象只有当在多节点环境下才能最大的发挥它的价值前面所介绍的Kafka配置方式都是基于单节点的配置由于本文主要是为了系统的梳理一下Kafka的配置及使用因此对于配置这一块不再花费大的篇幅去详细介绍如果需要到多节点配置Kafka可以自行查阅其他资料。
