详解Python的Twisted框架中reactor事件管理器的用法

铺垫
在大量的实践中，似乎我们总是通过类似的方式来使用异步编程：

• 监听事件
• 事件发生执行对应的回调函数
• 回调完成（可能产生新的事件添加进监听队列）
• 回到1，监听事件

因此我们将这样的异步模式称为Reactor模式，例如在iOS开发中的Run Loop概念，实际上非常类似于Reactor loop，主线程的Run Loop监听屏幕UI事件，一旦发生UI事件则执行对应的事件处理代码，还可以通过GCD等方式产生事件至主线程执行。

上图是boost对Reactor模式的描绘，Twisted的设计就是基于这样的Reactor模式，Twisted程序就是在等待事件、处理事件的过程中不断循环。

from twisted.internet import reactor
reactor.run()

</div>

reactor是Twisted程序中的单例对象。

reactor
reactor是事件管理器，用于注册、注销事件，运行事件循环，当事件发生时调用回调函数处理。关于reactor有下面几个结论:

• Twisted的reactor只有通过调用reactor.run()来启动。
• reactor循环是在其开始的进程中运行，也就是运行在主进程中。
• 一旦启动，就会一直运行下去。reactor就会在程序的控制下（或者具体在一个启动它的线程的控制下）。
• reactor循环并不会消耗任何CPU的资源。
• 并不需要显式的创建reactor，只需要引入就OK了。

最后一条需要解释清楚。在Twisted中，reactor是Singleton（也就是单例模式），即在一个程序中只能有一个reactor，并且只要你引入它就相应地创建一个。上面引入的方式这是twisted默认使用的方法，当然了，twisted还有其它可以引入reactor的方法。例如，可以使用twisted.internet.pollreactor中的系统调用来poll来代替select方法。

若使用其它的reactor，需要在引入twisted.internet.reactor前安装它。下面是安装pollreactor的方法：

from twisted.internet import pollreactor
pollreactor.install()

</div>

如果你没有安装其它特殊的reactor而引入了twisted.internet.reactor，那么Twisted会根据操作系统安装默认的reactor。正因为如此，习惯性做法不要在最顶层的模块内引入reactor以避免安装默认reactor，而是在你要使用reactor的区域内安装。
下面是使用 pollreactor重写上上面的程序:

from twited.internet import pollreactor
pollreactor.install()
from twisted.internet import reactor
reactor.run()

</div>

那么reactor是如何实现单例的？来看一下from twisted.internet import reactor做了哪些事情就并明白了。

下面是twisted/internet/reactor.py的部分代码:

# twisted/internet/reactor.py
import sys
del sys.modules['twisted.internet.reactor']
from twisted.internet import default
default.install()

</div>

注：Python中所有加载到内存的模块都放在sys.modules，它是一个全局字典。当import一个模块时首先会在这个列表中查找是否已经加载了此模块，如果加载了则只是将模块的名字加入到正在调用import的模块的命名空间中。如果没有加载则从sys.path目录中按照模块名称查找模块文件，找到后将模块载入内存，并加入到sys.modules中，并将名称导入到当前的命名空间中。

假如我们是第一次运行from twisted.internet import reactor，因为sys.modules中还没有twisted.internet.reactor，所以会运行reactory.py中的代码，安装默认的reactor。之后，如果导入的话，因为sys.modules中已存在该模块，所以会直接将sys.modules中的twisted.internet.reactor导入到当前命名空间。

default中的install：

# twisted/internet/default.py
def _getInstallFunction(platform):
  """
  Return a function to install the reactor most suited for the given platform.

  @param platform: The platform for which to select a reactor.
  @type platform: L{twisted.python.runtime.Platform}

  @return: A zero-argument callable which will install the selected
    reactor.
  """
  try:
    if platform.isLinux():
      try:
        from twisted.internet.epollreactor import install
      except ImportError:
        from twisted.internet.pollreactor import install
    elif platform.getType() == 'posix' and not platform.isMacOSX():
      from twisted.internet.pollreactor import install
    else:
      from twisted.internet.selectreactor import install
  except ImportError:
    from twisted.internet.selectreactor import install
  return install


install = _getInstallFunction(platform)

</div>

很明显，default中会根据平台获取相应的install。Linux下会首先使用epollreactor，如果内核还不支持，就只能使用pollreactor。Mac平台使用pollreactor，windows使用selectreactor。每种install的实现差不多，这里我们抽取selectreactor中的install来看看。

# twisted/internet/selectreactor.py：
def install():
  """Configure the twisted mainloop to be run using the select() reactor.
  """
  # 单例
  reactor = SelectReactor()
  from twisted.internet.main import installReactor
  installReactor(reactor)

# twisted/internet/main.py：
def installReactor(reactor):
  """
  Install reactor C{reactor}.

  @param reactor: An object that provides one or more IReactor* interfaces.
  """
  # this stuff should be common to all reactors.
  import twisted.internet
  import sys
  if 'twisted.internet.reactor' in sys.modules:
    raise error.ReactorAlreadyInstalledError("reactor already installed")
  twisted.internet.reactor = reactor
  sys.modules['twisted.internet.reactor'] = reactor

</div>

在installReactor中，向sys.modules添加twisted.internet.reactor键，值就是再install中创建的单例reactor。以后要使用reactor，就会导入这个单例了。

SelectReactor
# twisted/internet/selectreactor.py
@implementer(IReactorFDSet)
class SelectReactor(posixbase.PosixReactorBase, _extraBase)

</div> implementer表示SelectReactor实现了IReactorFDSet接口的方法，这里用到了zope.interface，它是python中的接口实现，有兴趣的同学可以去看下。</div>

IReactorFDSet接口主要对描述符的获取、添加、删除等操作的方法。这些方法看名字就能知道意思，所以我就没有加注释。

# twisted/internet/interfaces.py
class IReactorFDSet(Interface):

  def addReader(reader):

  def addWriter(writer):

  def removeReader(reader):

  def removeWriter(writer):

  def removeAll():

  def getReaders():

  def getWriters():
reactor.listenTCP()

</div>

示例中的reactor.listenTCP()注册了一个监听事件，它是父类PosixReactorBase中方法。

# twisted/internet/posixbase.py
@implementer(IReactorTCP, IReactorUDP, IReactorMulticast)
class PosixReactorBase(_SignalReactorMixin, _DisconnectSelectableMixin,
            ReactorBase):

  def listenTCP(self, port, factory, backlog=50, interface=''):
    p = tcp.Port(port, factory, backlog, interface, self)
    p.startListening()
    return p

# twisted/internet/tcp.py
@implementer(interfaces.IListeningPort)
class Port(base.BasePort, _SocketCloser):
  def __init__(self, port, factory, backlog=50, interface='', reactor=None):
    """Initialize with a numeric port to listen on.
    """
    base.BasePort.__init__(self, reactor=reactor)
    self.port = port
    self.factory = factory
    self.backlog = backlog
    if abstract.isIPv6Address(interface):
      self.addressFamily = socket.AF_INET6
      self._addressType = address.IPv6Address
    self.interface = interface
  ...

  def startListening(self):
    """Create and bind my socket, and begin listening on it.
     创建并绑定套接字，开始监听。

    This is called on unserialization, and must be called after creating a
    server to begin listening on the specified port.
    """
    if self._preexistingSocket is None:
      # Create a new socket and make it listen
      try:
        # 创建套接字
        skt = self.createInternetSocket()
        if self.addre