Producer/Consumer

Kafka的生产者（Producer）和消费者（Consumer）的关系，可以通过一个餐厅的例子来形象地说明。

1. 餐厅的故事

想象一个忙碌的餐厅，这里有：

• 厨师（Producers）：负责准备美味的菜肴。
• 服务台（Kafka Topic）：菜肴准备好后，厨师会将它们放到服务台上，服务台有多个部分，每部分代表一个不同类型的菜（即Kafka中的不同Partition）。
• 服务员（Consumers）：负责从服务台上取走菜肴，并将它们送到顾客手中。

在这个餐厅中，有时候会有特别多的订单，厨师需要快速高效地准备菜肴。每当一道菜准备好，他们就会把它放到对应的部分在服务台上。服务台非常长，可以容纳很多菜肴，让不同的服务员能够同时服务多个顾客，提高效率。

2. Kafka的工作方式

• Producers（厨师）：在Kafka中，生产者的角色是发布消息到Topic中。就像厨师准备好菜肴后，会将它们放到服务台的对应部分。
• Kafka Topic（服务台）：Topic是消息的分类，可以细分为多个Partitions（服务台的多个部分），这样可以提高并行处理的能力。每个Partition都是一个独立的队列。
• Consumers（服务员）：消费者从Topic中读取消息。如果有多个消费者在同一个Consumer Group中，它们可以像一队服务员那样协作，每个人负责从服务台的一部分取菜，这样可以更快地服务所有顾客。每个消费者负责读取特定Partition中的消息，确保每条消息都能被及时处理。

3. 生动的场景

假设一天晚上，餐厅接到了一个大型宴会的预订，需要同时准备多道菜。这时，厨师们（Producers）开始忙碌起来，每准备好一道菜，就会放到服务台（Topic）的指定位置（Partition）。服务员们（Consumers）各自负责一部分服务台，快速地将菜肴送到顾客手中。

在这个过程中，如果某一部分的菜准备得特别快，服务台上的这一部分就会堆积更多的菜肴。负责这一部分的服务员需要加快速度，以确保所有的菜肴都能及时送出。这就像在Kafka中，如果某个Partition的消息积压，负责这个Partition的消费者就需要更快地处理消息，以防止延迟。

通过这个例子，我们可以看到，Kafka的Producer和Consumer之间是如何通过Topic（服务台）和Partition（服务台的不同部分）协作的，以实现高效、可靠的消息处理。

4. 测试Demo

4.1 KafkaProducer

from kafka import KafkaProducer, KafkaConsumer
from kafka.errors import kafka_errors
import traceback
import json
import timeimport logginglogging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')def process():# Kafka配置,需自行修改bootstrap_servers = ['ip:port']producer_topic = 'XXX_topic'# Kafka生产者producer = KafkaProducer(bootstrap_servers=bootstrap_servers,value_serializer=lambda m: json.dumps(m).encode('utf-8'),api_version=(1,0,0))data = {"task_id": 1,"image_path": "XXX","video_path": "XXX","guidence_text": "XXX",}# Kafka请求监听try:res = data# 发送结果到Kafkaproducer.send(producer_topic, res)logging.info(f"send data to {producer_topic}")time.sleep(3)except Exception as e:# 记录错误日志logging.error(f"Error processing kafka request: {e}")if __name__ == "__main__":process()

4.2 KafkaConsumer

from kafka import KafkaConsumer
import json
import logginglogging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')def consume_messages():# Kafka配置bootstrap_servers = ['ip:port']consumer_topic = 'XXX'consumer_group = 'XXX'# Kafka消费者consumer = KafkaConsumer(consumer_topic,bootstrap_servers=bootstrap_servers,group_id=consumer_group,# auto_offset_reset='earliest',  # 从最早的消息开始读取auto_offset_reset= "latest",value_deserializer=lambda m: json.loads(m.decode('utf-8'))  # 解码JSON格式的消息)logging.info(f"Started consuming messages from {consumer_topic}")# 消费消息try:for message in consumer:msg = message.valuelogging.info(f"Received message: {msg}")print(f"msg:{msg}")except KeyboardInterrupt:logging.info("Stopping consumer...")except Exception as e:logging.error(f"Error while consuming messages: {e}")finally:consumer.close()if __name__ == "__main__":consume_messages()