我正在使用Python3.7和confluent kafka。
下面是我用来轮询kafka服务器并读取消息的伪代码。
while True:
MSG = CONSUMER.poll(0.1)
if MSG is None:
CONSUMER.commit()
print('No msg')
continue
if MSG.error():
print("Consumer error: {}".format(MSG.error()))
CONSUMER.commit()
continue
try:
rawMsg = format(MSG.value().decode('utf-8'))
testmsg = json.loads(rawMsg)
except:
print('invalid json format msg')
CONSUMER.commit()
如果kafka服务器由于某种原因关闭或断开连接,我希望抛出一个异常。
当前,如果发生上述情况,while循环将继续运行,不会出现任何错误,也不会打印任何消息。
如何获取异常或检查kafka服务器是否每次都可以在循环中连接(如果需要进行一些检查,那么它应该是轻量级的)。
暂无答案!
目前还没有任何答案,快来回答吧!