php网站有哪些/seo工具有哪些
安装依赖
在开始之前，需要安装 kafka-python 库。可以通过以下命令安装：
pip install kafka-python
创建生产者
生产者负责将消息发送到 Kafka 主题。以下是一个简单的生产者示例：
from kafka import KafkaProducer
import json
import time# 配置 Kafka 服务器地址和序列化方式
producer = KafkaProducer(bootstrap_servers='****:9092', # Kafka Broker 地址value_serializer=lambda v: json.dumps(v).encode('utf-8'), # JSON 序列化acks='all', # 确保消息被所有副本确认retries=3 # 失败重试次数
)# 发送单条消息
producer.send('testTopic', # 目标主题value={'message': 'Hello Kafka', 'timestamp': int(time.time())}
)# 批量发送消息（示例发送10条）
messages = [{'id': i, 'data': f'Message {i}'} for i in range(10)]
for msg in messages:producer.send('testTopic', value=msg)time.sleep(0.1) # 控制发送频率防止阻塞# 确保所有消息发送完成并关闭连接
producer.flush(timeout=10)
producer.close()
创建消费者
消费者负责从 Kafka 主题中读取消息。以下是一个简单的消费者示例：
from kafka import KafkaConsumer
import json# 配置消费者组和反序列化方式
consumer = KafkaConsumer('testTopic', # 订阅主题bootstrap_servers='****:9092',group_id='my_consumer_group', # 消费者组（同一组共享消息）auto_offset_reset='earliest', # 从最早未消费的消息开始读取value_deserializer=lambda x: json.loads(x.decode('utf-8')) # JSON 反序列化
)# 持续消费消息
try:for message in consumer:print(f"""收到消息:主题: {message.topic}分区: {message.partition}偏移量: {message.offset}内容: {message.value}""")
except KeyboardInterrupt:print("用户中断操作")
finally:consumer.close() # 关闭消费者连接Kafka 服务器
运行 Kafka 服务器，快速启动 Kafka 和 Zookeeper：
bin/kafka-server-start.sh -daemon config/server.properties创建指定topic的kafka的分区
./kafka-topics.sh --alter --topic testTopic --partitions 2 --bootstrap-server localhost:9092
查询指定topic的kafka分区
./kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic testTopic通过以上步骤，可以快速构建一个基于 Kafka 的消息队列系统，并在 Python 中实现消息的生产和消费。
注意：
1、kafka广播模式：
不同的分组（group_id参数不同）收到的数据是一致的，类似于广播模式
2、kafka路由模式：
同一分组（group_id参数相同）对同一批数据进行处理，如果kafka服务器的分区数量大于该分组内的消费者数量，则每个消费者可以分到一些分区，每个消费者去处理自己对应分区里面的数据，类似于路由模式下的多消费者情形，如果消费者数量大于分区数，则多出来的消费者是被闲置的