kafka IllegalStateError：生产者被强制关闭答案

【问题标题】：kafka IllegalStateError: Producer is closed forcefullykafka IllegalStateError：生产者被强制关闭
【发布时间】：2019-01-26 05:55:12
【问题描述】：

我有以下代码：

 __kafka_client = KafkaProducer(value_serializer=lambda m: json.dumps(m).encode('ascii'))
 for(msg in messages):    
    __kafka_client.send('TOPIC', dumps(msg)).add_callback(on_kafka_send_success).add_errback(on_kafka_send_error)

def on_kafka_send_success(record_metadata):
    print(record_metadata.topic)
def on_kafka_send_error(excp):
   kafa_error_message  = "Error while sending kafka message :"+str(excp);
   raise Exception(excp)

` 循环结束时出现以下错误：

Error while sending kafka message :IllegalStateError: Producer is closed forcefully.
NoneType: None

如何避免/解决此问题？

【问题讨论】：

标签： python python-3.x apache-kafka kafka-producer-api

【解决方案1】：

解决了

__kafka_client = KafkaProducer(value_serializer=lambda m: json.dumps(m).encode('ascii'))
 for(msg in messages):    
    __kafka_client.send('TOPIC', dumps(msg)).add_callback(on_kafka_send_success).add_errback(on_kafka_send_error)

__kafka_clinet.flush()

【讨论】：

【解决方案2】：

在调用生产/发送方法后立即关闭客户端进程时，此问题会重现。显然，在解释器完成 send() 调用之前，不一定要刷新消息缓冲区。就像在 Bhuneshwer 的回答 (https://stackoverflow.com/a/51940821/12912120) 中一样，您可以使用 flush() 在终止进程之前刷新消息缓冲区，或者如果您不立即结束进程，消息将很快发送（这个“足够快”可以通过 queue.buffering 配置。最大毫秒）。

请注意，在缓冲区刷新完成之前，flush() 会一直阻塞，因此在每次调用产生后使用刷新等效于实现同步流，并且可能会影响性能。

请参阅 confluent-kafka 人员的这个答案以获取更多信息： https://github.com/confluentinc/confluent-kafka-python/issues/137#issuecomment-282427382

【讨论】：