首次消费后Kafka Consumer无法接收消息问题求助

核心问题分析 #

从代码来看，导致Consumer卡住无法接收新消息的核心原因是Kafka生产者/消费者实例重复创建且未正确关闭，引发offset提交异常、资源泄漏；其次是消息处理逻辑的细节缺失，导致消息传递不顺畅。

具体修复点 #

1. 复用Kafka实例，避免重复创建 #

每次调用消费/生产函数时都新建Kafka实例，且未关闭，会导致Kafka集群认为旧实例仍在运行，offset无法正确提交，新实例无法拉取最新消息。同时重复创建实例会浪费资源，消息也可能因未flush延迟发送。

修复Engine的生产者/消费者

全局初始化一次实例，复用至程序结束，确保offset正常提交：

# 全局初始化，避免重复创建
producer = KafkaProducer(value_serializer=lambda v: pickle.dumps(v),
                         bootstrap_servers=ip_kafka + ":" + puerto_kafka)

consumer = KafkaConsumer("movimiento", value_deserializer=lambda v: pickle.loads(v),
                         bootstrap_servers=ip_kafka + ':' + puerto_kafka,
                         group_id="engine", auto_offset_reset="latest", 
                         enable_auto_commit=True, consumer_timeout_ms=10000)

def consume_pos_act():
    NUEVAS_POSICIONES = []
    contador = 0

    for mensaje in consumer:
        men = mensaje.value
        info = men.split(',')
        NUEVAS_POSICIONES.append(info)
        print("Me han pasado:", info)

        contador += 1
        if contador == len(LISTA_DRONES):
            consumer.commit()  # 手动提交offset，标记已消费消息
            break

    return NUEVAS_POSICIONES

def produce_pos_act(nuevas_posiciones):
    for pos in nuevas_posiciones:
        print("Enviando a dron", pos)
        producer.send("dron", value=pos)
    producer.flush()  # 确保所有消息写入Kafka

# 程序结束时统一关闭资源
# producer.close()
# consumer.close()

修复AD_Drone的生产者/消费者

在Drone类初始化时创建实例，复用至任务结束：

class AD_Drone:
    def __init__(self, ...):
        # 初始化时创建一次实例
        self.PRODUCER = KafkaProducer(value_serializer=lambda v: pickle.dumps(v),
                                     bootstrap_servers=self.IP_KAFKA + ":" + self.PORT_KAFKA)
        self.CONSUMER = KafkaConsumer("dron", value_deserializer=lambda v: pickle.loads(v),
                                     bootstrap_servers=self.IP_KAFKA + ":" + self.PORT_KAFKA, 
                                     group_id=self.ALIAS, auto_offset_reset="latest", 
                                     consumer_timeout_ms=10000)
        # 其他初始化逻辑...

    def run(self):
        # 先获取最终位置，再进入循环
        self.consume_pos('f')
        while not self.COMPLETADO:
            print("Pasando a producir pos actual")
            self.produce_pos_act()
            print("Pasando a consumir pos actual")
            sleep(2)
            self.consume_pos('a')
        # 任务结束关闭资源
        self.PRODUCER.close()
        self.CONSUMER.close()

    def produce_pos_act(self):
        pos_actual = self.calcula_pos_actual(int(self.X_FIN), int(self.Y_FIN), int(self.X_ACTUAL), int(self.Y_ACTUAL))
        dato = str(self.ID) + ',' + str(pos_actual)
        print("Envío a Engine", dato)
        self.PRODUCER.send("movimiento", value=dato)
        self.PRODUCER.flush()  # 确保消息立即发送

    def consume_pos(self, pos):
        for mensaje in self.CONSUMER:
            print("Entro en consumición")
            datos = mensaje.value
            datos = datos.split(',')
            print("Recibo", datos)

            if datos[0] == self.ID:
                if pos == 'a':
                    self.X_ACTUAL = datos[1]
                    self.Y_ACTUAL = datos[2]
                    print(f"Soy {self.ALIAS} mi posición ACTUAL es ({self.X_ACTUAL},{self.Y_ACTUAL})")
                elif pos == 'f':
                    self.X_FIN = datos[1]
                    self.Y_FIN = datos[2]
                    print(f"Soy {self.ALIAS} mi posición FINAL es ({self.X_FIN},{self.Y_FIN})")
                self.CONSUMER.commit()  # 提交offset，避免重复消费
                break

2. 调整Engine首次消息发送逻辑 #

首次发送无人机最终位置时，确保消息全部写入Kafka，并给无人机足够接收时间：

# Enviando posiciones finales definitivas
for dron in LISTA_DRONES:
    producer.send("dron", value=str(dron.ID) + "," + str(dron.X_FIN) + ',' + str(dron.Y_FIN))
producer.flush()  # 统一发送所有消息
sleep(3)  # 给无人机预留接收时间

3. 修复Drone启动逻辑 #

原代码中Drone未获取最终位置就开始计算移动方向，会导致逻辑错误。需先调用consume_pos('f')获取终点坐标，再进入生产-消费循环（如上述run方法所示）。

其他优化建议 #

• 替换pickle序列化：pickle存在安全风险，跨语言兼容性差，建议改用JSON序列化（json.dumps/json.loads）。
• 替换print为日志模块：使用logging记录关键操作，便于排查问题。
• 增加异常处理：在消息反序列化、位置计算等环节添加try-except块，避免程序崩溃。

内容的提问来源于stack exchange，提问作者Saúl Campello Mas