如何在Python的server.py中访问sendmsg函数内部的inputData数据

• 作用域使用错误：sendmsg是嵌套在time协程中的内部函数，你在sendmsg里定义的inputData默认是局部变量，global关键字只对全局作用域变量生效，对嵌套函数的外层作用域变量无效，所以外部拿不到值。
• 阻塞逻辑冲突：mqttc.loop_forever()是同步阻塞方法，运行后会直接占满线程，导致asyncio的websocket事件循环完全被卡住，后续所有逻辑都无法执行。
• 缺少变量暴露入口：原来的代码没有对外提供获取inputData的通道，也没有MQTT消息触发后的执行钩子。

完整可运行server.py代码 #

import asyncio
import datetime
import websockets
import json
import bcrypt
import paho.mqtt.client as mqtt
import threading

# 全局存储MQTT触发后生成的inputData，初始为None表示还未触发
latest_input_data = None
# 加读写锁避免多线程变量冲突
data_lock = threading.Lock()
websocketData = {}

def sendmsg(arr, usr_details, hashCode, websocket):
    print("mqtt msg", arr)
    inputData = {
        "type": "dataSend",
        "name": usr_details['name'],
        "accessKey": hashCode.decode('utf-8'),
        "senderKey": arr[0],
        "msg": arr[1]
    }
    printData = {
        "name": usr_details['name'],
        "msg": arr[1]
    }
    # 更新全局变量
    global latest_input_data
    with data_lock:
        latest_input_data = inputData
    # 如果需要收到MQTT消息后主动推给websocket客户端，可以取消下面这行的注释
    # asyncio.run(websocket.send(json.dumps(printData)))
    print("inputData已更新", latest_input_data)

async def time(websocket, path):
    usr_details = await websocket.recv()
    usr_details = json.loads(usr_details)
    usr_name = usr_details['name']
    print(usr_details)    

    usr_name = usr_details['name']
    now = datetime.datetime.utcnow().isoformat() + "Z"
    salt = bcrypt.gensalt(5)
    hashCode = bcrypt.hashpw(bytes(usr_name, 'utf-8'), salt=salt)

    websocketData[hashCode.decode('utf-8')] = websocket
    print(websocketData)

    data = {"name": usr_name, "date": now, "hashCodeOfUser": (hashCode).decode('UTF-8'), "msg": "User Created"}
    print(data)
    await websocket.send(json.dumps(data))

    ############################################################
    broker="broker.emqx.io"

    def on_message(mqttc, obj, msg):
        message = (msg.payload).decode("utf-8")
        arr = message.split()
        sendmsg(arr, usr_details, hashCode, websocket)

    mqttc = mqtt.Client(client_id="")
    mqttc.on_message = on_message
    mqttc.connect(broker, 1883, 60)
    mqttc.subscribe("vinay/websockets/topic/1",qos=0)
    # 用非阻塞的loop_start代替阻塞的loop_forever，不影响asyncio运行
    mqttc.loop_start()
    ###############################################################

    # 后续你要访问inputData的逻辑可以写在这里，示例：
    while True:
        await asyncio.sleep(1)
        with data_lock:
            if latest_input_data is not None:
                print("外部拿到最新inputData：", latest_input_data)
                # 这里可以加你要执行的其他逻辑
                # 处理完可以重置为None，等下一次MQTT触发再更新
                # latest_input_data = None


start_server = websockets.serve(time, "127.0.0.1", 5678 )
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()

客户端代码（和原有逻辑兼容，无需修改） #

<!DOCTYPE html>
<html>
<head>
    <!-- <meta http-equiv="refresh" content="30"> -->
    <title>WebSocket demo</title>
</head>
<body>
    <ul id="received_messages"></ul>
    <script>
        const user_credentials = {
            type: "credentials",
            name: "vinay",
            email: "vinay5678@gmail.com"
        }
        let userHashCode = ''
        const receivedMessages = document.getElementById("received_messages");

        var ws = new WebSocket("ws://127.0.0.1:5678/"),
            messages = document.createElement('ul');

        ws.onopen = function() {
            ws.send(JSON.stringify(user_credentials));
        };

        ws.onmessage = function(event) {
            let message = document.createElement('li');
            console.log(event.data)
            message.innerText = `${event.data}`;
            document.body.appendChild(message);
        };
    </script>
</body>
</html>

1. 每次MQTT lens推送消息后，latest_input_data会自动更新，只有推送过消息这个变量才会有值，初始状态为None
2. 你在任何需要访问inputData的位置，加锁读取latest_input_data即可，判断变量不为None就说明已经有MQTT消息触发
3. 如果需要每收到一次MQTT消息只处理一次，可以处理完手动把latest_input_data重置为None

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