Python之Kafka安装与使用

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

• zookeeper-server-start.bat ..\..\config/zookeeper.properties
  
• kafka-server-start.bat ..\..\config/server.properties
  

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

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

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

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

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)

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)

运行如下代码暂时没有显示，此时开另一个程序运行生产者代码，即会有输出显示；即监听生产者发布的消息，只要生产者新产生了一条消息，他就会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)))