认识twisted reactor

Posted 2021-01-15 mactec

一、reactor是单线程模型，简单粗暴，也就是说网络IO和我们的业务逻辑一般是在一个线程里，其中网络IO通过event loop的方式去异步执行，效率也很高。看下官网的这幅图，比较清晰

　　

　　twisted主要帮我们处理的事情就是去监听socket，等待事件完成触发回调，然后回到我们的业务逻辑中。

　　可以看出，由于业务逻辑和网络异步IO都在reactor线程里，当业务逻辑比较耗时（如复杂计算，SQL慢）的时候，twisted帮不到我们，需要我们自己去解决，如用线程池去跑费时任务，防止阻塞reactor主线程。

 

二、reactor提供了较丰富的方法供我们调用，常用的如下：

　　reactor.callInThread： 可以新起一个线程，用来处理耗时逻辑，需注意线程安全

　　reactor.callFromThread： 和上面相反，在别的线程调用，去回到reactor主线程执行逻辑

　　threads.deferToThread： 将任务放到线程池去执行 以免阻塞主线程，特点是返回的是deferred对象，这样我们可以加入回调逻辑

　　reactor.suggestThreadPoolSize: 设置线程池线程数

　　reactor.run/stop 启动、停止主线程

　　reactor.callLater 延迟执行

　　reactor.callFromRunning 立刻执行

 

三、 看下小例子

　　

# -*- coding:utf-8 –*-
import threading
import time
from twisted.internet import reactor, threads

thread = threading.current_thread()


def leaveReactor():
    print "[%s]等待3秒,当前线程：%s"%(nowtime(),threading.current_thread().getName())
    time.sleep(3)
    reactor.callFromThread(backReactor)

def backReactor():
    print "[%s]回到reactor，线程：%s"%(nowtime(),threading.current_thread().getName())

def deferReactor():
    time.sleep(3)
    return "asyJob"

def handle_success(x):
    print "[%s]回调成功 :%s "%(nowtime(),x)

def work():
    reactor.callInThread(leaveReactor)
    d = threads.deferToThread(deferReactor)
    d.addCallback(handle_success)
    print "[%s] 主线程：%s"%(nowtime(),threading.current_thread().getName())

def nowtime():
    return time.strftime(‘%Y-%m-%d,%X‘, time.localtime())

if __name__ == ‘__main__‘:
    reactor.suggestThreadPoolSize(8)
    reactor.callWhenRunning(work)
    reactor.run()