婚恋网站设计,兰州企业网络优化服务,网站建设虚拟空间,担路网如何快速做网站为了向Kafka集群生产和消费消息#xff0c;我们可以使用confluent-kafka库#xff0c;它是Confluent为Python提供的官方Kafka客户端。以下是一个简化的示例#xff0c;展示如何将Kafka的生产者和消费者操作封装到一个类中#xff1a;
首先#xff0c;确保你已经安装了所需…为了向Kafka集群生产和消费消息我们可以使用confluent-kafka库它是Confluent为Python提供的官方Kafka客户端。以下是一个简化的示例展示如何将Kafka的生产者和消费者操作封装到一个类中
首先确保你已经安装了所需的库
pip install confluent-kafka 然后你可以使用以下代码
from confluent_kafka import Producer, Consumer, KafkaErrorclass KafkaManager:def __init__(self, bootstrap_servers):self.bootstrap_servers bootstrap_serversdef produce(self, topic, key, value):生产消息到Kafkap Producer({bootstrap.servers: self.bootstrap_servers})def delivery_report(err, msg):Called once for each message produced to indicate delivery result.if err is not None:print(Message delivery failed: {}.format(err))else:print(Message delivered to {} [{}].format(msg.topic(), msg.partition()))p.produce(topic, keykey, valuevalue, callbackdelivery_report)p.flush()def consume(self, topic, group_id, timeout1.0):从Kafka消费消息c Consumer({bootstrap.servers: self.bootstrap_servers,group.id: group_id,auto.offset.reset: earliest})c.subscribe([topic])while True:msg c.poll(timeout)if msg is None:continueif msg.error():if msg.error().code() KafkaError._PARTITION_EOF:print(Reached end of partition)else:print(Error while consuming message: {}.format(msg.error()))else:print(Received message: {}.format(msg.value().decode(utf-8)))c.close()# 使用示例
if __name__ __main__:manager KafkaManager(localhost:9092)# 生产消息manager.produce(test_topic, key1, value1)# 消费消息manager.consume(test_topic, test_group)
pip install kafka-python
from kafka import KafkaProducer, KafkaConsumerclass KafkaManager:def __init__(self, bootstrap_servers):self.bootstrap_servers bootstrap_serversdef produce(self, topic, key, value):生产消息到Kafkaproducer KafkaProducer(bootstrap_serversself.bootstrap_servers,key_serializerstr.encode,value_serializerstr.encode)producer.send(topic, keykey, valuevalue)producer.flush()producer.close()def consume(self, topic, group_id, timeout10):从Kafka消费消息consumer KafkaConsumer(topic,bootstrap_serversself.bootstrap_servers,group_idgroup_id,auto_offset_resetearliest,key_deserializerbytes.decode,value_deserializerbytes.decode)for message in consumer:print(fReceived message: {message.value})consumer.close()# 使用示例
if __name__ __main__:manager KafkaManager(localhost:9092)# 生产消息manager.produce(test_topic, key1, value1)# 消费消息manager.consume(test_topic, test_group)
