如何在Twisted的LoopingCall中处理含time.sleep的阻塞函数

当然可以继续用LoopingCall，核心是让你的work()方法返回Deferred，把阻塞逻辑放到线程中执行——Twisted的LoopingCall支持处理返回Deferred的任务，会自动等待Deferred完成后再调度下一次执行，不会阻塞reactor。

修改后的实现代码 #

from twisted.internet import threads, reactor
from twisted.internet.task import LoopingCall
import time

# 假设elements是预先定义的列表
elements = [1, 2, 3]

looping_call = LoopingCall(work)
looping_call.start(10)


def work():
    # 把整个阻塞的循环逻辑放到线程中执行
    return threads.deferToThread(_blocking_work_loop)


def _blocking_work_loop():
    result = False
    for i in elements:
        if blocking_function(i):  # 保持顺序处理的逻辑不变
            result = True

    if result:
        # 如果dosomething是阻塞操作，同样要放到线程里；如果是异步操作可直接调用
        dosomething()
    return result


def blocking_function(i):
    time.sleep(5)
    return True  # or False


def dosomething():
    print("检测到符合条件的元素，执行操作")


if __name__ == "__main__":
    reactor.run()

关键说明 #

• threads.deferToThread会把传入的函数放到单独的线程中执行，返回一个Deferred对象，Twisted的reactor会在线程完成后触发回调。
• LoopingCall检测到work()返回Deferred时，会暂停下一次调度，直到这个Deferred被回调（即线程里的阻塞逻辑执行完毕），再按照设定的间隔（10秒）启动下一次任务。
• 如果你只需要把单个blocking_function放到线程，也可以在循环里逐个用deferToThread，但这样会变成并发执行；而你需要顺序处理，所以把整个循环包装到一个线程里更符合需求。
• 注意：线程内的代码不要直接操作Twisted的reactor或者其他线程不安全的组件，如果需要在阻塞逻辑完成后触发异步操作，可以在Deferred的回调里处理。

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