在 Python 中使用装饰器来封装 Kafka 的操作类（如 `KafkaProducer` 和 `KafkaConsumer`）是一种良好的设计模式，可以提升代码的可维护性和复用性。通过装饰器，可以将 Kafka 的连接管理、消息发送与消费逻辑解耦，使业务代码更加简洁和专注。 以下是一个使用装饰器封装 KafkaProducer 和 KafkaConsumer 的示例： ### 使用装饰器封装 KafkaProducer ```python from functools import wraps from kafka import KafkaProducer import json # 定义一个装饰器用于封装KafkaProducer的初始化和关闭 def kafka_producer_wrapper(bootstrap_servers='localhost:9092', value_serializer=lambda v: json.dumps(v).encode('utf-8')): def decorator(func): @wraps(func) def wrapper(*args, **kwargs): producer = KafkaProducer(bootstrap_servers=bootstrap_servers, value_serializer=value_serializer) try: result = func(producer, *args, **kwargs) producer.flush() return result finally: producer.close() return wrapper return decorator # 使用装饰器定义一个发送消息的函数 @kafka_producer_wrapper(bootstrap_servers='localhost:9092') def send_message(producer, topic, message): future = producer.send(topic, value=message) return future.get(timeout=10) # 等待发送结果 # 调用函数发送消息 send_message('my_topic', {'key': 'value'}) ``` ### 使用装饰器封装 KafkaConsumer ```python from functools import wraps from kafka import KafkaConsumer import json # 定义一个装饰器用于封装KafkaConsumer的初始化和关闭 def kafka_consumer_wrapper(bootstrap_servers='localhost:9092', group_id='my-group', auto_offset_reset='earliest', value_deserializer=lambda m: json.loads(m.decode('utf-8'))): def decorator(func): @wraps(func) def wrapper(*args, **kwargs): consumer = KafkaConsumer( bootstrap_servers=bootstrap_servers, group_id=group_id, auto_offset_reset=auto_offset_reset, value_deserializer=value_deserializer ) try: result = func(consumer, *args, **kwargs) return result finally: consumer.close() return wrapper return decorator # 使用装饰器定义一个消费消息的函数 @kafka_consumer_wrapper(bootstrap_servers='localhost:9092', group_id='my-group') def consume_messages(consumer, topic): consumer.subscribe([topic]) for message in consumer: print(f"Received message: {message.value}") # 可以在这里处理消息逻辑 # 调用函数消费消息 consume_messages('my_topic') ``` ### 说明 - **装饰器设计**：上述代码中定义了两个装饰器 `kafka_producer_wrapper` 和 `kafka_consumer_wrapper`，分别用于封装 `KafkaProducer` 和 `KafkaConsumer` 的初始化和关闭逻辑。 - **参数传递**：装饰器接受 Kafka 的配置参数（如 `bootstrap_servers`、`group_id` 等），并在内部创建相应的 Kafka 客户端实例。 - **资源管理**：通过 `try...finally` 确保 Kafka 客户端实例在使用完毕后正确关闭，避免资源泄露。 - **业务逻辑分离**：业务函数只需关注消息的发送或消费逻辑，而无需关心底层的连接管理和异常处理。 这种封装方式不仅提高了代码的可读性和可维护性，还使得 Kafka 操作更加模块化，便于在不同项目中复用。