Kafka生产者处理与代理失去连接

当Kafka生产者与代理失去连接时，可以通过以下方式解决：

1. 重试机制：在生产者代码中添加重试机制，当发生连接断开的情况时，重试发送消息直到成功连接到代理。以下是一个使用重试机制的示例代码：

from kafka import KafkaProducer
from kafka.errors import KafkaError

# Kafka连接配置
bootstrap_servers = 'localhost:9092'
topic = 'my_topic'

# 创建生产者
producer = KafkaProducer(bootstrap_servers=bootstrap_servers)

# 发送消息
def send_message(message):
    try:
        producer.send(topic, message)
        producer.flush()
        print("Message sent successfully")
    except KafkaError as e:
        print("Failed to send message:", e)
        # 连接断开，进行重试
        producer.close()
        producer = KafkaProducer(bootstrap_servers=bootstrap_servers)
        send_message(message)

# 调用发送消息函数
send_message("Hello Kafka")

2. 使用Kafka的回调函数：Kafka生产者可以设置回调函数，在消息发送成功或失败时调用该函数。在回调函数中可以处理连接断开的情况并进行重试。以下是一个使用回调函数的示例代码：

from kafka import KafkaProducer
from kafka.errors import KafkaError

# Kafka连接配置
bootstrap_servers = 'localhost:9092'
topic = 'my_topic'

# 创建生产者
producer = KafkaProducer(bootstrap_servers=bootstrap_servers)

# 发送消息的回调函数
def on_send_success(record_metadata):
    print("Message sent successfully")

def on_send_error(excp):
    print("Failed to send message:", excp)
    # 连接断开，进行重试
    producer.close()
    producer = KafkaProducer(bootstrap_servers=bootstrap_servers)
    send_message(message)

# 发送消息
def send_message(message):
    try:
        producer.send(topic, message).add_callback(on_send_success).add_errback(on_send_error)
        producer.flush()
    except Exception as e:
        print("Failed to send message:", e)
        # 连接断开，进行重试
        producer.close()
        producer = KafkaProducer(bootstrap_servers=bootstrap_servers)
        send_message(message)

# 调用发送消息函数
send_message("Hello Kafka")

这些代码示例展示了两种处理Kafka生产者与代理失去连接的方法。可以根据实际需求选择适合的方法来解决问题。