詳解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()

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()

如果你沒有安裝其它特殊的reactor而引入了twisted.internet.reactor，那麼Twisted會根據作業系統安裝預設的reactor。正因為如此，習慣性做法不要在最頂層的模組內引入reactor以避免安裝預設reactor，而是在你要使用reactor的區域內安裝。
下面是使用 pollreactor重寫上上面的程式:

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

那麼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()

註：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)

   

很明顯，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

   

在installReactor中，在sys.modules中加入twisted.internet.reactor鍵，值就是再install建立的單例reactor。以後要使用reactor，就會導入這個單例了。

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

   

implementer表示SelectReactor實現了IReactorFDSet介面的方法，這裡用到了zope.interface，它是python中的介面可以實現，有興趣的同學可以去看下。

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()

   

範例中的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.addressFamily == socket.AF_INET6:
          addr = _resolveIPv6(self.interface, self.port)
        else:
          addr = (self.interface, self.port)
        # 绑定
        skt.bind(addr)
      except socket.error as le:
        raise CannotListenError(self.interface, self.port, le)
      # 监听
      skt.listen(self.backlog)
    else:
      # Re-use the externally specified socket
      skt = self._preexistingSocket
      self._preexistingSocket = None
      # Avoid shutting it down at the end.
      self._shouldShutdown = False
 
    # Make sure that if we listened on port 0, we update that to
    # reflect what the OS actually assigned us.
    self._realPortNumber = skt.getsockname()[1]
 
    log.msg("%s starting on %s" % (
        self._getLogPrefix(self.factory), self._realPortNumber))
 
    # The order of the next 5 lines is kind of bizarre. If no one
    # can explain it, perhaps we should re-arrange them.
    self.factory.doStart()
    self.connected = True
    self.socket = skt
    self.fileno = self.socket.fileno
    self.numberAccepts = 100
 
    # startReading调用reactor的addReader方法将Port加入读集合
    self.startReading()

   

整个逻辑很简单，和正常的server端一样，创建套接字、绑定、监听。不同的是将套接字的描述符添加到了reactor的读集合。那么假如有了client连接过来的话，reactor会监控到，然后触发事件处理程序。

reacotr.run()事件主循环

# twisted/internet/posixbase.py
@implementer(IReactorTCP, IReactorUDP, IReactorMulticast)
class PosixReactorBase(_SignalReactorMixin, _DisconnectSelectableMixin,
            ReactorBase)
 
# twisted/internet/base.py
class _SignalReactorMixin(object):
 
  def startRunning(self, installSignalHandlers=True):
    """
    PosixReactorBase的父类_SignalReactorMixin和ReactorBase都有该函数，但是
    _SignalReactorMixin在前，安装mro顺序的话，会先调用_SignalReactorMixin中的。
    """
    self._installSignalHandlers = installSignalHandlers
    ReactorBase.startRunning(self)
 
  def run(self, installSignalHandlers=True):
    self.startRunning(installSignalHandlers=installSignalHandlers)
    self.mainLoop()
 
  def mainLoop(self):
    while self._started:
      try:
        while self._started:
          # Advance simulation time in delayed event
          # processors.
          self.runUntilCurrent()
          t2 = self.timeout()
          t = self.running and t2
          # doIteration是关键，select,poll,epool实现各有不同
          self.doIteration(t)
      except:
        log.msg("Unexpected error in main loop.")
        log.err()
      else:
        log.msg('Main loop terminated.')

mianLoop就是最终的主循环了，在循环中，调用doIteration方法监控读写描述符的集合，一旦发现有描述符准备好读写，就会调用相应的事件处理程序。

# twisted/internet/selectreactor.py
@implementer(IReactorFDSet)
class SelectReactor(posixbase.PosixReactorBase, _extraBase):
 
  def __init__(self):
    """
    Initialize file descriptor tracking dictionaries and the base class.
    """
    self._reads = set()
    self._writes = set()
    posixbase.PosixReactorBase.__init__(self)
 
  def doSelect(self, timeout):
    """
    Run one iteration of the I/O monitor loop.
 
    This will run all selectables who had input or output readiness
    waiting for them.
    """
    try:
      # 调用select方法监控读写集合，返回准备好读写的描述符
      r, w, ignored = _select(self._reads,
                  self._writes,
                  [], timeout)
    except ValueError:
      # Possibly a file descriptor has gone negative?
      self._preenDescriptors()
      return
    except TypeError:
      # Something *totally* invalid (object w/o fileno, non-integral
      # result) was passed
      log.err()
      self._preenDescriptors()
      return
    except (select.error, socket.error, IOError) as se:
      # select(2) encountered an error, perhaps while calling the fileno()
      # method of a socket. (Python 2.6 socket.error is an IOError
      # subclass, but on Python 2.5 and earlier it is not.)
      if se.args[0] in (0, 2):
        # windows does this if it got an empty list
        if (not self._reads) and (not self._writes):
          return
        else:
          raise
      elif se.args[0] == EINTR:
        return
      elif se.args[0] == EBADF:
        self._preenDescriptors()
        return
      else:
        # OK, I really don't know what's going on. Blow up.
        raise
 
    _drdw = self._doReadOrWrite
    _logrun = log.callWithLogger
    for selectables, method, fdset in ((r, "doRead", self._reads),
                      (w,"doWrite", self._writes)):
      for selectable in selectables:
        # if this was disconnected in another thread, kill it.
        # ^^^^ --- what the !@#*? serious! -exarkun
        if selectable not in fdset:
          continue
        # This for pausing input when we're not ready for more.
 
        # 调用_doReadOrWrite方法
        _logrun(selectable, _drdw, selectable, method)
 
  doIteration = doSelect
 
  def _doReadOrWrite(self, selectable, method):
    try:
      # 调用method，doRead或者是doWrite，
      # 这里的selectable可能是我们监听的tcp.Port
      why = getattr(selectable, method)()
    except:
      why = sys.exc_info()[1]
      log.err()
    if why:
      self._disconnectSelectable(selectable, why, method=="doRead")

那么假如客户端有连接请求了，就会调用读集合中tcp.Port的doRead方法。

# twisted/internet/tcp.py
 
@implementer(interfaces.IListeningPort)
class Port(base.BasePort, _SocketCloser):
 
  def doRead(self):
    """Called when my socket is ready for reading.
    当套接字准备好读的时候调用
 
    This accepts a connection and calls self.protocol() to handle the
    wire-level protocol.
    """
    try:
      if platformType == "posix":
        numAccepts = self.numberAccepts
      else:
        numAccepts = 1
      for i in range(numAccepts):
        if self.disconnecting:
          return
        try:
          # 调用accept
          skt, addr = self.socket.accept()
        except socket.error as e:
          if e.args[0] in (EWOULDBLOCK, EAGAIN):
            self.numberAccepts = i
            break
          elif e.args[0] == EPERM:
            continue
          elif e.args[0] in (EMFILE, ENOBUFS, ENFILE, ENOMEM, ECONNABORTED):
            log.msg("Could not accept new connection (%s)" % (
              errorcode[e.args[0]],))
            break
          raise
 
        fdesc._setCloseOnExec(skt.fileno())
        protocol = self.factory.buildProtocol(self._buildAddr(addr))
        if protocol is None:
          skt.close()
          continue
        s = self.sessionno
        self.sessionno = s+1
        # transport初始化的过程中，会将自身假如到reactor的读集合中，那么当它准备
        # 好读的时候，就可以调用它的doRead方法读取客户端发过来的数据了
        transport = self.transport(skt, protocol, addr, self, s, self.reactor)
        protocol.makeConnection(transport)
      else:
        self.numberAccepts = self.numberAccepts+20
    except:
      log.deferr()

doRead方法中，调用accept产生了用于接收客户端数据的套接字，将套接字与transport绑定，然后把transport加入到reactor的读集合。当客户端有数据到来时，就会调用transport的doRead方法进行数据读取了。

Connection是Server（transport实例的类）的父类，它实现了doRead方法。

# twisted/internet/tcp.py
@implementer(interfaces.ITCPTransport, interfaces.ISystemHandle)
class Connection(_TLSConnectionMixin, abstract.FileDescriptor, _SocketCloser,
         _AbortingMixin):
 
  def doRead(self):
    try:
      # 接收数据
      data = self.socket.recv(self.bufferSize)
    except socket.error as se:
      if se.args[0] == EWOULDBLOCK:
        return
      else:
        return main.CONNECTION_LOST
 
    return self._dataReceived(data)
 
  def _dataReceived(self, data):
    if not data:
      return main.CONNECTION_DONE
    # 调用我们自定义protocol的dataReceived方法处理数据
    rval = self.protocol.dataReceived(data)
    if rval is not None:
      offender = self.protocol.dataReceived
      warningFormat = (
        'Returning a value other than None from %(fqpn)s is '
        'deprecated since %(version)s.')
      warningString = deprecate.getDeprecationWarningString(
        offender, versions.Version('Twisted', 11, 0, 0),
        format=warningFormat)
      deprecate.warnAboutFunction(offender, warningString)
    return rval

   

_dataReceived中调用了示例中我们自定义的EchoProtocol的dataReceived方法处理数据。

至此，一个简单的流程，从创建监听事件，到接收客户端数据就此结束了。

更多詳解Python的Twisted框架中reactor事件管理器的用法相关文章请关注PHP中文网！