使用confluent_kafka实现Python脚本跨网通信遇阻，求解决方案

问题背景 #

尝试用confluent_kafka实现两个Python脚本的跨网通信，流程为：

• Script1向Script2发送温度读取请求
• Script2接收请求后返回当前温度给Script1并打印
  但现有代码稳定性差，多数情况下Script2无法消费到消息，需确认该方案可行性及替代方案。

现有代码 #

Script1.py #

from confluent_kafka import Consumer, Producer
from dotenv import load_dotenv
import os

load_dotenv(".env")

# 从.env文件加载敏感数据
bootstrap_server = os.getenv("BOOTSTRAP_SERVER")
sasl_user_name = os.getenv("CLIENT_ID")
sasl_password = os.getenv("CLIENT_SECRET")

# 配置Kafka生产者
p = Producer({
      'bootstrap.servers': bootstrap_server,
      'security.protocol': 'SASL_SSL',
      'sasl.mechanisms': 'PLAIN',
      'sasl.username': sasl_user_name,
      'sasl.password': sasl_password,
})

# 配置Kafka消费者
c = Consumer({
    'bootstrap.servers': bootstrap_server,
    'security.protocol': 'SASL_SSL',
    'sasl.mechanisms': 'PLAIN',
    'sasl.username': sasl_user_name,
    'sasl.password': sasl_password,
    'group.id': 'script1-group',
    'enable.auto.commit': False,
    'auto.offset.reset': 'latest',
    
})

def delivery_report(err, msg):
    if err is not None:
        print('消息投递失败: {}'.format(err))
    else:
        print('消息投递至 {} [{}]'.format(msg.topic(), msg.partition()))

# 发送温度请求消息
p.poll(0)
data = 'temperature'
p.produce('script2', data.encode('utf-8'), callback=delivery_report)
p.flush()

# 订阅返回结果的topic
c.subscribe(['script1'])

x = True
while x == True:
    msg = c.poll(1.0)
    if msg is None:
        continue
    if msg.error():
        print("消费者错误: {}".format(msg.error()))
        continue
    temp = msg.value().decode('utf-8')
    print("当前温度为 " + temp)
    x = False

Script2.py #

from confluent_kafka import Consumer, Producer
from dotenv import load_dotenv
import os

load_dotenv(".env")

# 从.env文件加载敏感数据
bootstrap_server = os.getenv("BOOTSTRAP_SERVER")
sasl_user_name = os.getenv("CLIENT_ID")
sasl_password = os.getenv("CLIENT_SECRET")

# 配置Kafka生产者
p = Producer({
      'bootstrap.servers': bootstrap_server,
      'security.protocol': 'SASL_SSL',
      'sasl.mechanisms': 'PLAIN',
      'sasl.username': sasl_user_name,
      'sasl.password': sasl_password,
})

# 配置Kafka消费者
c = Consumer({
    'bootstrap.servers': bootstrap_server,
    'security.protocol': 'SASL_SSL',
    'sasl.mechanisms': 'PLAIN',
    'sasl.username': sasl_user_name,
    'sasl.password': sasl_password,
    'group.id': 'script2Group',
    'enable.auto.commit': False,
    'auto.offset.reset': 'latest',
    
})

def delivery_report(err, msg):
    if err is not None:
        print('消息投递失败: {}'.format(err))
    else:
        print('消息投递至 {} [{}]'.format(msg.topic(), msg.partition()))

# 订阅请求消息的topic
c.subscribe(['script2'])

x = True
while x == True:
    msg = c.poll(1.0)
    if msg is None:
        continue
    if msg.error():
        print("消费者错误: {}".format(msg.error()))
        continue
    if msg.value().decode('utf-8') == 'temperature':
        p.poll(0)
        data = "20 C"
        p.produce('script1', data.encode('utf-8'), callback=delivery_report)
        print("已向Script1发送温度数据")
        p.flush()
        x = False

代码问题分析 #

1. 消息消费时机问题：Script1先发送消息，若Script2启动滞后，由于消费者配置auto.offset.reset: 'latest'，只会消费启动后的新消息，之前发送的请求会被错过。
2. 偏移量未手动提交：配置enable.auto.commit: False但未在消费后手动调用c.commit()，会导致消费者重启后重复消费或丢失消息。
3. 无重试机制：Script2的消费循环只运行一次就退出，若首次poll未获取到消息，直接终止程序，没有重试逻辑。
4. topic存在性未验证：若script2或script1 topic未提前创建，消息可能无法正常存储和消费。

修复建议（使Kafka方案可行） #

1. 调整启动顺序：先启动Script2，确保消费者已订阅topic后，再启动Script1发送请求。
2. 修改消费者配置：将auto.offset.reset改为'earliest'，确保消费者能获取topic中历史消息。
3. 手动提交偏移量：在Script2成功消费并处理消息后，添加c.commit()提交偏移量；Script1同理。
4. 增加重试逻辑：Script2的消费循环不要仅运行一次，可设置超时时间或持续监听，避免因网络延迟错过消息。
5. 提前创建topic：确保Kafka集群中已存在script1和script2两个topic，可通过Kafka命令行或管理工具创建。

修改后的Script2关键部分示例：

import time

x = True
timeout = 30  # 设置30秒超时
start_time = time.time()
while x == True:
    if time.time() - start_time > timeout:
        print("超时未收到请求消息")
        x = False
        continue
    msg = c.poll(1.0)
    if msg is None:
        continue
    if msg.error():
        print("消费者错误: {}".format(msg.error()))
        continue
    if msg.value().decode('utf-8') == 'temperature':
        p.poll(0)
        data = "20 C"
        p.produce('script1', data.encode('utf-8'), callback=delivery_report)
        print("已向Script1发送温度数据")
        p.flush()
        c.commit(msg)  # 手动提交偏移量
        x = False

替代方案推荐 #

1. MQTT #

• 适合轻量级跨网设备/脚本通信，低带宽消耗，支持发布订阅模式，自带QoS保证消息可靠性。
• 可使用paho-mqtt库实现，部署MQTT broker（如EMQX、Mosquitto）即可跨网通信。

2. gRPC #

• 基于HTTP/2的高性能RPC框架，天生支持请求-响应模式，适合需要同步通信的场景，自带序列化和错误处理。
• 定义proto文件即可生成客户端和服务端代码，跨语言兼容，适合复杂数据交互。

3. RabbitMQ（AMQP） #

• 提供可靠的消息队列机制，支持多种消息模式（点对点、发布订阅），自带消息确认、重试机制，稳定性高。
• 使用pika库实现，适合需要确保消息不丢失的场景。

4. HTTP请求 #

• 最简单直接的方式，Script2启动一个HTTP服务（如用Flask或FastAPI），Script1发送GET/POST请求获取温度数据。
• 无需额外消息中间件，部署成本低，适合单次请求响应的简单场景。

内容的提问来源于stack exchange，提问作者MichaelLong102