RecordAccumulator is closed

2019-12-03 19:04发布

问题:

我在使用python-kafka,发送数据的时候,通过线程调用多个producer,但是运行的时候报了这个错我,RecordAccumulator is closed,尝试了几种方法还是没有解决,不知道这个是什么原因导致的,
def produce(path):
producer.send(topic,message,key)
if key_log:
for i in range(num1):
t1 = threading.Thread(target=produce,args=(producer,path_list1[i],key_log,))
t1.start()
if key_msg:
for i in range(num2):
t2 = threading.Thread(target=produce,args=(producer,path_list2[i],key_msg,))
t2.start()

代码大概就是这样的意思,里面详细的我没写了,我尝试在t1.start()后面,join一下就可以,但是不join(),就出现上面的错误,但是join之后下面一个for循环的线程就没运行了

回答1:

这个问题主要是因为我开启子线程后,主线程没有等待子线程运行结束,就直接结束掉主线程,关闭了程序,因为我在这段代码的下面调用了
producer.flush()
producer.close()
这样就会导致我虽然每次启动了程序,线程也调用了一次,但是运行了一次就被迫关闭了,所以加了join()就可以运行,但是只能运行t1,因为t2一直被阻塞在,所以我改为在producer.flush()这句代码前加一个死循环,让程序一直阻塞在在上面,即
while True:
time.sleep(3600)
这样之后就可以运行了,只不过死循环后面的代码好像一直不能运行,就算想结束也是手动结束。