PythonAnywhere上Flask MQTT IoT服务端运行异常求助

本地在Visual Studio运行正常的Flask-MQTT代码，部署到PythonAnywhere后出现以下问题：

• 已配置MQTT凭证、禁用TLS、设置5秒心跳，通过mqtt_client = Mqtt(app1)完成实例化
• @mqtt_client.on_connect()回调完全未触发（无日志输出），导致消息发布、on_message功能全部失效
• 但在PythonAnywhere的bash控制台手动导入MQTT实例并调用发布函数，能正常连接代理并发送消息
• 尝试过两种初始化方式：先实例化Mqtt()再调用mqtt_client.init_app(app1)，以及调用mqtt.run()，均未解决问题

原始代码 #

app1.config['MQTT_BROKER_URL'] = 'mybroker'
app1.config['MQTT_BROKER_PORT'] = 1883
app1.config['MQTT_USERNAME'] = ' '
app1.config['MQTT_PASSWORD'] = ' '
app1.config['MQTT_KEEPALIVE'] = 5
app1.config['MQTT_TLS_ENABLED'] = False

mqtt_client = Mqtt(app1)

import sys as syss

@mqtt_client.on_connect()
def handle_connect(client,userdata,flags,rc):
   if rc == 0:
      print('Connected successfully', file = syss.stderr)
      mqtt_client.subscribe('esp/copra')
   else:
      print('Bad connection. Code:', rc , file=syss.stderr)

testmqtt="a"

@mqtt_client.on_message()
def handle_mqtt_message(client,userdata,message):
   payload = message.payload.decode()
   print("payload is " + payload)

@app1.route("/testmqtt")
def testmqtt():
   print("test mqtt here")
   mqtt_client.publish('esp/copra',"pythonanywhere")
   return{"mqtt": "mqtt"}

if name == "main":
   #app.config['SESSION_TYPE'] = 'filesystem'
   mqtt_client.init_app(app1)
   app1.run()

问题分析与解决方案 #

1. 修复重复初始化与语法错误 #

• 代码中同时使用mqtt_client = Mqtt(app1)和mqtt_client.init_app(app1)，重复初始化会导致回调绑定失效，移除main块中的mqtt_client.init_app(app1)。
• 修正if name == "main"为if __name__ == "__main__"，这是Python脚本入口的标准语法，错误写法会导致本地运行逻辑无法正确执行，且PythonAnywhere的WSGI环境不会执行这段代码。

2. 适配PythonAnywhere的WSGI运行模式 #

PythonAnywhere通过WSGI服务器加载Flask应用，不会执行if __name__ == "__main__"中的代码，因此需要确保MQTT客户端在应用加载阶段就完成初始化：

• 将MQTT初始化逻辑放在全局作用域，而非仅在脚本直接运行时执行。
• 检查PythonAnywhere的WSGI配置文件，确保正确导入Flask应用和MQTT客户端，示例配置：

  from your_app_filename import app1, mqtt_client
  application = app1
  

3. 启动MQTT后台循环线程 #

flask-mqtt依赖后台线程维护MQTT连接，在WSGI环境下需要手动启动该线程，避免连接无法建立：

# 初始化MQTT客户端后添加
mqtt_client.client.loop_start()

4. 调整日志输出方式 #

PythonAnywhere的Flask应用日志不会直接输出到sys.stderr，改用Flask内置日志系统便于排查问题：

import logging
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

@mqtt_client.on_connect()
def handle_connect(client,userdata,flags,rc):
    if rc == 0:
        logger.info('Connected successfully')
        mqtt_client.subscribe('esp/copra')
    else:
        logger.error(f'Bad connection. Code: {rc}')

修正后的完整代码 #

import logging
from flask import Flask
from flask_mqtt import Mqtt

app1 = Flask(__name__)

# MQTT配置
app1.config['MQTT_BROKER_URL'] = 'mybroker'
app1.config['MQTT_BROKER_PORT'] = 1883
app1.config['MQTT_USERNAME'] = ''  # 移除多余空格
app1.config['MQTT_PASSWORD'] = ''  # 移除多余空格
app1.config['MQTT_KEEPALIVE'] = 5
app1.config['MQTT_TLS_ENABLED'] = False

# 初始化MQTT客户端并启动循环线程
mqtt_client = Mqtt(app1)
mqtt_client.client.loop_start()

# 日志配置
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

@mqtt_client.on_connect()
def handle_connect(client,userdata,flags,rc):
    if rc == 0:
        logger.info('Connected successfully')
        mqtt_client.subscribe('esp/copra')
    else:
        logger.error(f'Bad connection. Code: {rc}')

@mqtt_client.on_message()
def handle_mqtt_message(client,userdata,message):
    payload = message.payload.decode()
    logger.info(f"payload is {payload}")

@app1.route("/testmqtt")
def testmqtt():
    logger.info("test mqtt here")
    mqtt_client.publish('esp/copra',"pythonanywhere")
    return{"mqtt": "mqtt"}

# 仅本地运行时执行
if __name__ == "__main__":
    app1.run(debug=True)

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