Kafka生产者：使用回调函数处理异步发送中的异常

下面是一个使用回调函数处理Kafka生产者异步发送中的异常的示例代码：

from kafka import KafkaProducer
from kafka.errors import KafkaError

def on_send_success(record_metadata):
    print('Message sent successfully to topic:', record_metadata.topic)
    print('Partition:', record_metadata.partition)
    print('Offset:', record_metadata.offset)

def on_send_error(excp):
    print('Message failed to send:', str(excp))

def produce_message(producer, topic, message):
    producer.send(topic, message).add_callback(on_send_success).add_errback(on_send_error)

# 创建Kafka生产者
producer = KafkaProducer(bootstrap_servers='localhost:9092')

# 发送消息
topic = 'my_topic'
message = 'Hello Kafka!'

produce_message(producer, topic, message)

# 关闭Kafka生产者
producer.close()

在上面的代码中，我们定义了两个回调函数：on_send_success和on_send_error。on_send_success函数会在消息成功发送后被调用，并输出发送的消息的元数据信息。on_send_error函数会在发送消息时发生异常时被调用，并输出异常信息。

然后，我们定义了一个produce_message函数，它接受一个Kafka生产者对象、一个主题和一个消息作为参数。在这个函数中，我们使用producer.send方法发送消息，并通过add_callback方法添加on_send_success回调函数，通过add_errback方法添加on_send_error回调函数。

最后，我们创建一个Kafka生产者对象，调用produce_message函数发送消息，并在发送完成后关闭生产者。

请确保在运行代码之前已经安装了kafka-python包。