Python之Kafka安装与使用_十一姐_python 安装kafka

大大的周 02-07 3845

目录一、kafka基础概念了解二、下载安装Kafka三、KafkaTool可视化工具四、python操作kafka1、生产者代码2、消费者代码

一、kafka基础概念了解 Kafka是一种高吞吐量、持久性、分布式的发布订阅的消息队列系统kafka文档为什么应该学习 KafkaKafka 里面的信息是如何被消费的？使用生成器把Kafka写入速度提高1000倍Kafka名词概念producer：消息生产者，向 Kafka Broker 发消息的客户端broker: 一台 Kafka 机器就是一个 broker。一个集群由多个 broker 组成。一个 broker 可以容纳多个 topictopic: 可以理解为一个队列，topic 将消息分类，生产者和消费者面向的是同一个 topicoffset:消费者消费的位置信息，监控数据消费到什么位置，当消费者挂掉，再重新恢复的时候，可以从消费位置继续消费partition: topic在物理上的分区，一个topic可以分为一个或多个分区，每个partition是一个有序的队列consumer: 消息消费者，从Kafka Broker 取消息的客户端consumer group: 消费者组（CG），消费者组内每个消费者，负责消费不同分区的数据，提高消费能力。一个分区只能由组内一个消费者消费，消费者组之间互不影响。所有的消费者都属于某个消费者组，即消费者组是逻辑上的一个订阅者Zookeeper: Kafka 集群能够正常工作，需要依赖于 zookeeper，zookeeper 帮助Kafka 存储和管理集群信息二、下载安装Kafka 1、点击下载kafka，点击图中链接里面的下载链接 2、然后将压缩包解压即可，然后我们到kafka_2.12-3.1.1\bin\windows目录下启动kafka 3、启动kafka前需要使用Zookeeper，如果没有安装Zookeeper，可以使用Kafka自带打包和配置好的Zookeeper。先切换到kafka_2.12-3.1.1\bin\windows目录下，如下图打开两个cmd窗口，分别执行如下两个命令，注意这两个窗口不要关 zookeeper-server-start.bat ..\..\config/zookeeper.properties kafka-server-start.bat ..\..\config/server.properties 然后python就可以操作kafka了，注意这两个窗口不关三、KafkaTool可视化工具

KafkaTool下载，然后一路next即可，中间的安装路径可以自定义

打开OffsetExploret，如图跳出提示，随意命名一个Cluster name即可连接

打开界面如下，可视化工具更详细的功能使用文档

每个message默认是以二进制形式展示的，可在如下Tools>Settings>Topics下设置为String的形式展示

四、python操作kafka pip install kafka-pythonkafka-python基本使用如何使用Python读写Kafka？如果中断了重启，对于同一个group id消费者会继续上次的终止点继续消费，如果换了group id则从头开始消费因为Kafka的broker是无状态的，所以consumer必须使用partition offset来记录消费了多少数据。如果一个consumer指定了一个topic的offset，意味着该consumer已经消费了该offset之前的所有数据；consumer可以通过指定offset，从topic的指定位置开始消费数据；consumer的offset存储在Zookeeper中，参考文档新版kafka中，默认的partition是50 1、生产者代码如图，第一个参数指定topic名称from kafka import KafkaProducer producer = KafkaProducer(bootstrap_servers=['localhost:9092']) record = {"title": "今天真美丽", "url": "http://123123.jewr.com"} future = producer.send('my_topic', value=bytes(f'{record}', encoding='utf-8'), partition=0) result = future.get(timeout=10) print(result) 参数value_serializer用来指定序列化的方式，这里我使用 json 来序列化数据，从而实现我向 Kafka 传入一个字典，Kafka 自动把它转成 JSON 字符串的效果from kafka import KafkaProducer import json producer = KafkaProducer( bootstrap_servers=['localhost:9092'], max_block_ms=10000, max_request_size=10485760, send_buffer_bytes=10485760, value_serializer=lambda m: json.dumps(m, ensure_ascii=False).encode(), api_version=(0, 10) ) for _id in range(200, 205): record = {"title": "今天真美丽", "_id": _id} future = producer.send(topic='my_topic', value=record) result = future.get(timeout=10) print(result) 2、消费者代码

运行如下代码暂时没有显示，此时开另一个程序运行生产者代码，即会有输出显示；即监听生产者发布的消息，只要生产者新产生了一条消息，他就会print输出msg；

topic：topic的名称

group_id：指定此消费者实例属于的组名，group_id这个参数后面的字符串可以任意填写；如果两个程序的Topic与group_id相同，那么他们读取的数据不会重复，两个程序的Topic相同，但是group_id不同，那么他们各自消费全部数据，互不影响

bootstrap_servers ：指定kafka服务器

auto_offset_rest 这个参数有两个值，earliest和latest，如果省略这个参数，那么默认就是latest；参考如何使用Python读写Kafka：auto_offset_reset的作用，是在你的 group 第一次运行，还没有 offset 的时候，给你设定初始的 offset。而一旦你这个 group 已经有 offset 了，那么auto_offset_reset这个参数就不会再起作用了

import time from kafka import KafkaConsumer consumer = KafkaConsumer( 'my_topic', group_id='test_id', bootstrap_servers=['localhost:9092'], auto_offset_reset='earliest' ) for msg in consumer: print(msg) print(f"topic = {msg.topic}") # topic default is string print(f"partition = {msg.partition}") print(f"value = {msg.value.decode()}") # bytes to string print(f"timestamp = {msg.timestamp}") print("time = ", time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(msg.timestamp / 1000)))