如何使用超时停止阻塞函数 subscribe.simple答案

【问题标题】：How to use timeout to stop blocking function subscribe.simple如何使用超时停止阻塞函数 subscribe.simple
【发布时间】：2019-08-06 13:43:55
【问题描述】：

我想用timeout来停止mqtt的阻塞功能，我用了一个timeout_decorator模块，它可以停止命令功能但不能停止阻塞功能，subscribe.simple。

以下代码运行成功

import time
import timeout_decorator

@timeout_decorator.timeout(5, timeout_exception=StopIteration)
def mytest():
    print("Start")
    for i in range(1,10):
        time.sleep(1)
        print("{} seconds have passed".format(i))

if __name__ == '__main__':
    mytest()

结果如下：

Start
1 seconds have passed
2 seconds have passed
3 seconds have passed
4 seconds have passed
Traceback (most recent call last):
  File "timeutTest.py", line 12, in <module>
    mytest()
  File "/home/gyf/.local/lib/python3.5/site-packages/timeout_decorator/timeout_decorator.py", line 81, in new_function
    return function(*args, **kwargs)
  File "timeutTest.py", line 8, in mytest
    time.sleep(1)
  File "/home/gyf/.local/lib/python3.5/site-packages/timeout_decorator/timeout_decorator.py", line 72, in handler
    _raise_exception(timeout_exception, exception_message)
  File "/home/gyf/.local/lib/python3.5/site-packages/timeout_decorator/timeout_decorator.py", line 45, in _raise_exception
    raise exception()
timeout_decorator.timeout_decorator.TimeoutError: 'Timed Out'

但我使用 subscribe.simple API 失败了

import timeout_decorator

@timeout_decorator.timeout(5)
def sub():
    # print(type(msg))
    print("----before simple")
    # threading.Timer(5,operateFail,args=)
    msg = subscribe.simple("paho/test/simple", hostname=MQTT_IP,port=MQTT_PORT,)
    print("----after simple")
    return msg


publish.single("paho/test/single", "cloud to device", qos=2, hostname=MQTT_IP,port=MQTT_PORT)
try:
    print("pub")
    msg = sub()
    print(msg)
except StopIteration as identifier:
    print("error")

结果无限等待

pub
----before simple

我希望包含 subscribe.simple API 的功能可以在 5 秒后停止。

【问题讨论】：

为什么一定要subscribe.simple()？为什么不使用普通的客户端订阅处理和单独的线程来跟踪待处理的请求？
因为我需要等待设备的响应才能在页面上显示操作是否成功。如果我使用多线程，主线程根据使用普通客户端的子线程的响应计划返回200或404响应将直接结束
你找到解决办法了吗？

标签： python-3.x synchronization mqtt paho

【解决方案1】：

Asyncio 将无法在同一线程中处理阻塞函数。因此使用asyncio.wait_for 失败。然而，受this blog post 的启发，我使用loop.run_in_executor 来保持对阻塞线程的控制。

from paho.mqtt import subscribe
import asyncio

MQTT_IP = "localhost"
MQTT_PORT = 1883
msg = None


def possibly_blocking_function():
    global msg
    print("listenning for message")
    msg = subscribe.simple(
        "paho/test/simple",
        hostname=MQTT_IP,
        port=MQTT_PORT,
    )
    print("message received!")


async def main():
    print("----before simple")
    try:
        await asyncio.wait_for(
            loop.run_in_executor(None, possibly_blocking_function), timeout=5
        )
    except asyncio.TimeoutError:
        pass
    print("----after simple")

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

输出：

----before simple
listenning for message
----after simple

请注意这并不完美，程序不会结束，因为有正在运行的任务。您可以使用各种解决方案退出它，但这超出了范围，因为我仍在寻找一种干净的方法来关闭该卡住的线程。

【讨论】：