部署为Web应用的Flask+MQTT客户端无法接收消息问题

核心问题定位 #

本地运行完全正常，但部署到PythonAnywhere后，集成的MQTT客户端无法接收/转发消息，而外部直接调用Flask接口正常。说明Flask服务本身无问题，问题出在MQTT客户端的连接、监听环节在PythonAnywhere环境下的适配性。

排查与解决步骤 #

1. 解决WSGI进程模型的限制 #

PythonAnywhere的Web应用基于WSGI运行，默认模式下无法维持MQTT所需的长期TCP连接——WSGI进程可能被重启、回收，导致MQTT监听线程中断。

• 解决方法：把MQTT客户端和Flask Web服务拆分，单独用PythonAnywhere的后台任务运行MQTT客户端：
  1. 新建独立的mqtt_client.py脚本存放MQTT逻辑
  2. 在PythonAnywhere控制台启动后台任务：nohup python3 mqtt_client.py &，确保进程持续运行

2. 补全日志排查连接状态 #

当前代码用print输出的日志在WSGI环境下无法查看，替换为文件日志定位问题：

import logging
logging.basicConfig(filename='mqtt_logs.log', level=logging.INFO, format='%(asctime)s - %(message)s')

@mqtt.on_connect()
def on_connect(client, userdata, flags, rc):
    logging.info(f"MQTT连接结果码：{rc}")
    client.subscribe(topic)

@mqtt.on_message()
def on_message(client, userdata, message):
    logging.info("收到MQTT消息")
    # 后续处理逻辑...

部署后在PythonAnywhere的Files页面查看mqtt_logs.log，确认是否成功连接Broker、是否收到消息。

3. 调整TLS配置兼容性 #

PythonAnywhere的Python环境中，ssl.PROTOCOL_TLS_CLIENT可能存在版本兼容问题，尝试替换为更通用的协议，并补充CA证书路径：

import certifi
import ssl

# 修改TLS相关配置
app.config['MQTT_TLS_VERSION'] = ssl.PROTOCOL_TLSv1_2
app.config['MQTT_TLS_CA_CERTS'] = certifi.where()  # 使用系统默认CA证书

先执行pip install certifi安装依赖库。

4. 修正内部POST请求路径 #

on_message中调用Flask接口的相对路径可能失效，替换为完整的内部/外部URL：

# 示例：使用PythonAnywhere的外部域名
response = requests.post('https://yourusername.pythonanywhere.com/api/core2/sensors/air-quality', 
                         data=data, headers=headers)
# 或本地回环地址（仅同一服务器内有效）
response = requests.post('http://localhost:5000/api/core2/sensors/air-quality', 
                         data=data, headers=headers)

5. 添加MQTT重连机制 #

防止Broker断开后无法自动恢复连接：

@mqtt.on_disconnect()
def on_disconnect(client, userdata, rc):
    logging.info(f"MQTT断开连接，结果码：{rc}")
    client.reconnect()

# 初始化时配置心跳和重连参数
app.config['MQTT_KEEPALIVE'] = 60  # 心跳间隔60秒
app.config['MQTT_RECONNECT_MAX_RETRIES'] = 5  # 最大重连次数

拆分后的MQTT客户端示例代码 #

import time
import ssl
import requests
import certifi
from flask import Flask
from flask_mqtt import Mqtt
from openapi_core import Info, OpenAPI
from http import HTTPStatus

info = Info(title='AIR API', version='1.0.0')
app = OpenAPI(__name__, info=info, template_folder='templates')

# MQTT配置
app.config['MQTT_BROKER_URL'] = "boker.url"
app.config['MQTT_BROKER_PORT'] = 8883
app.config['MQTT_USERNAME'] = "username"
app.config['MQTT_PASSWORD'] = "password"
app.config['MQTT_CLIENT_ID'] = "clientReceive" + str(time.time())
app.config['MQTT_TLS_ENABLED'] = True
app.config['MQTT_TLS_CA_CERTS'] = certifi.where()
app.config['MQTT_TLS_CERTFILE'] = None
app.config['MQTT_TLS_KEYFILE'] = None
app.config['MQTT_TLS_CERT_REQS'] = ssl.CERT_NONE
app.config['MQTT_TLS_VERSION'] = ssl.PROTOCOL_TLSv1_2
app.config['MQTT_KEEPALIVE'] = 60

topic = "mytopic"
mqtt = Mqtt(app)

@mqtt.on_connect()
def on_connect(client, userdata, flags, rc):
    print(f"Connected with result code {rc}")
    client.subscribe(topic)

@mqtt.on_disconnect()
def on_disconnect(client, userdata, rc):
    print(f"Disconnected with result code {rc}")
    client.reconnect()

@mqtt.on_message()
def on_message(client, userdata, message):
    data = message.payload.decode()
    headers = {'Content-Type': 'application/json'}
    try:
        response = requests.post('https://yourusername.pythonanywhere.com/api/core2/sensors/air-quality', 
                                 data=data, headers=headers)
        if response.status_code == HTTPStatus.CREATED:
            print('Data received and stored')
            print(f"Data: {data}")
        else:
            print(f'Error storing data, status code: {response.status_code}')
    except Exception as e:
        print(f"failed to forward message to Flask endpoint: {e}")

if __name__ == '__main__':
    app.run(debug=False, use_reloader=False)

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