### Python3 中 Kafka 消费者封装方法 在 Python3 中，可以利用 `confluent-kafka` 或 `kafka-python` 库来实现 Kafka 消费者的功能。为了提高代码的可重用性和模块化程度，通常会通过类的方式对消费者逻辑进行封装。 以下是基于 `kafka-python` 的一种常见封装方式： #### 封装 Kafka 消费者的核心逻辑 可以通过定义一个名为 `KafkaConsumerWrapper` 的类来封装消费者的初始化、订阅主题以及消息处理等功能[^1]。 ```python from kafka import KafkaConsumer class KafkaConsumerWrapper: def __init__(self, bootstrap_servers, group_id, auto_offset_reset='earliest'): """ 初始化 Kafka 消费者实例。 :param bootstrap_servers: Kafka 集群地址列表，例如 'localhost:9092' :param group_id: 消费组 ID :param auto_offset_reset: 偏移量重置策略，默认为 earliest """ self.consumer = KafkaConsumer( bootstrap_servers=bootstrap_servers, group_id=group_id, auto_offset_reset=auto_offset_reset ) def subscribe_topics(self, topics): """ 订阅指定的主题列表。 :param topics: 主题名称列表，例如 ['topic1', 'topic2'] """ self.consumer.subscribe(topics) def consume_messages(self, message_handler=None): """ 开始消费消息并调用回调函数处理每条消息。 :param message_handler: 处理单条消息的回调函数，默认打印日志 """ try: for message in self.consumer: if message_handler: message_handler(message) else: print(f"Received message: {message.value.decode('utf-8')}") except Exception as e: print(f"Error while consuming messages: {e}") def close(self): """关闭消费者连接""" self.consumer.close() ``` #### 使用封装后的消费者 下面是一个简单的例子，展示如何使用上述封装好的消费者类[^2]。 ```python def handle_message(message): """自定义的消息处理函数""" print(f"Custom handler received: {message.value.decode('utf-8')}") if __name__ == "__main__": # 创建消费者实例 consumer_wrapper = KafkaConsumerWrapper( bootstrap_servers="localhost:9092", group_id="my-group-id" ) # 订阅主题 consumer_wrapper.subscribe_topics(["test-topic"]) # 开始消费消息 consumer_wrapper.consume_messages(handle_message) ``` 这种封装方式不仅提高了代码的清晰度和可维护性，还允许开发者轻松扩展其他功能，比如错误处理、偏移管理等。 ---