Python の Twisted フレームワークのコア機能の分析

1. リアクター
の核心はリアクターであり、リアクターに関して言えば、最初の章の Dave の概念紹介では、同期/非同期の境界について言及することが避けられません。同期/非同期、ブロッキング/非ブロッキングについては、Zhihu のディスカッションで見つけることができます。プロクター（活性化剤）とリアクター（反応器）については、こちらのブログが詳しく紹介されているのでおすすめです。
リアクター モードのネットワーク IO に関する限り、非同期 IO ではなく同期 IO である必要があります。 Dave の最初の章で述べた非同期性の核心は、オペレーティング システムによってランダムに停止されるのではなく、タスクの制御を明示的に放棄することです。プログラマはタスクを順番に編成し、小さなステップを交互に実行して完了する必要があります。したがって、タスクの 1 つが別のタスクの出力を使用する場合、依存するタスク (つまり、出力を受け取るタスク) は、すべてを一度に受け取るのではなく、一連のビットまたはフラグメントを受け取るように設計する必要があります。
明示的かつ積極的にタスクの制御を放棄することは、Reactor をコルーチンのスケジューラーとみなすことができるコルーチンの考え方に似ています。 Reactor はイベント ループです。関心のあるイベント (読み取り/書き込み可能であるソケットなど) とプロセッサ (読み取りおよび書き込み操作の実行など) をリアクターに登録できます。イベントが発生すると、リアクターはプロセッサをコールバックします。プロセッサが実行された後、これはコルーチンのハング (yield) と同等であり、リアクタのイベント ループに戻り、次のイベントが来るのを待ってコールバックします。リアクター自体には同期イベント デマルチプレクサーがあり、select/epoll やその他のメカニズムによって実装できます。もちろん、ツイスト リアクターのイベント トリガーは必ずしも IO に基づく必要はなく、タイマーなどの他のメカニズムによってトリガーすることもできます。
ツイスト リアクターは、イベントやコールバック関数を積極的に登録する必要はありませんが、ポリモーフィズムを通じて実装されます (特定のクラスを継承し、対象のイベント インターフェイスを実装して、それをツイスト リアクターに渡します)。ツイスト リアクターに関しては、注意すべき点がいくつかあります。
twisted.internet.reactor はシングルトン モードであり、各プログラムはリアクターを 1 つだけ持つことができます。
リアクター コールバック関数内でできるだけ早く操作を完了するように努めてください。リアクターの本質はシングルスレッドです。特定のコールバック関数でブロックすると、リアクターのイベント ループ全体が常にブロックされます。は、reactor.stop() によって明示的に停止されますが、一般に、reactor.stop() を呼び出すことは、アプリケーションが終了することを意味します。

2. Twisted は使いやすいです。便利な高レベルの抽象化) を使用してリアクターを使用します:

# 示例一 twisted底层API的使用
from twisted.internet import reacto
from twisted.internet import main
from twisted.internet.interfaces import IReadDescriptor
import socket
 
class MySocket(IReadDescriptor):
  def __init__(self, address):
    # 连接服务器
    self.address = address
    self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    self.sock.connect(address)
    self.sock.setblocking(0)
 
    # tell the Twisted reactor to monitor this socket for reading
    reactor.addReader(self)
  
 # 接口: 告诉reactor 监听的套接字描述符
  def fileno(self):
    try:
      return self.sock.fileno()
    except socket.error:
      return -1
       
 # 接口: 在连接断开时的回调
  def connectionLost(self, reason):
    self.sock.close()
 
    reactor.removeReader(self)
  
 # 当应用程序需要终止时 调用:
    # reactor.stop()
 
 # 接口: 当套接字描述符有数据可读时
  def doRead(self):
    bytes = ''
 
 # 尽可能多的读取数据
    while True:
      try:
        bytesread = self.sock.recv(1024)
        if not bytesread:
          break
        else:
          bytes += bytesread
      except socket.error, e:
        if e.args[0] == errno.EWOULDBLOCK:
          break
        return main.CONNECTION_LOST
 
    if not bytes: 
      return main.CONNECTION_DONE
    else:
      # 在这里解析协议并处理数据
      print bytes

例 1 では、ツイスト リアクターの本質が明確にわかります。リスニング記述子を追加し、読み取り可能/書き込み可能なイベントをリッスンし、イベントが来たときに関数をコールバックし、引き続きリッスンします。コールバックが完了した後のイベント。

注意:

ソケットはノンブロッキングです。ブロッキングしている場合、リアクターの意味が失われます
IReadDescriptorを継承することでリアクターに必要なインターフェースを提供します
reactor.addReaderを通じてリアクターのリスニングオブジェクトにソケットクラスを追加します
main.CONNECTION_LOSTこれらの値を使用して、次のコールバックをある程度制御できます (イベントのシミュレーションと同様)。ただし、上記の MySocket クラスには主に次の欠点があります。フレームワークではなく、データが例外の読み取りと処理に役立ちます
ネットワーク IO とデータ処理は分離されずに混在しています

3. ツイストでは、リアクターに基づいてより高度な抽象化が確立されています。ネットワーク接続に関して、Twisted は次の 3 つの概念を確立しました:
トランスポート: ネットワーク接続層、ネットワーク接続とバイト データの読み取り/書き込みのみを担当します。
プロトコル: プロトコル層、サービス ビジネス関連のネットワーク プロトコル、バイト ストリームを必要なデータに変換します。アプリケーション用

プロトコル ファクトリ: プロトコルの作成を担当するプロトコル ファクトリ。各ネットワーク接続には Protocols オブジェクトがあります (プロトコル解析ステータスが保存されるため)。これらのツイストの概念は、Erlang の牧場ネットワーク フレームワークと牧場に非常に似ています。フレームワークは、新しいネットワーク接続があるときに、Ranch の起動時にトランスポートとプロトコルを自動的に作成し、ranch_protocol の動作を実装するモジュールです。 、これを受信します。対応するトランスポートに接続します。これにより、プロトコルでバイト ストリーム データを処理し、プロトコルに従ってデータを解析して処理できるようになります。同時に、データはトランスポート経由で送信できます (ranch はすでにバイト ストリーム データを読み取っています)。

ranch と同様に、Twisted もプロトコルを作成し、新しい接続が到着したときにトランスポートを渡します。dataReceived(self, data) インターフェイスでバイト ストリーム データを処理するだけで済みます。できる。現時点では、Twisted はネットワーク IO において真の非同期であると見なすことができ、ネットワーク IO と考えられる例外に対処し、ネットワーク IO とデータ処理を分離してトランスポートとプロトコルに抽象化し、プログラムの効率を向上させます。そして堅牢性。

# 示例二 twisted抽象的使用
from twisted.internet import reactor
from twisted.internet.protocol import Protocol, ClientFactory
class MyProtocol(Protocol):
  
 # 接口: Protocols初始化时调用，并传入Transports
 # 另外 twisted会自动将Protocols的factory对象成员设为ProtocolsFactory实例的引用
 #   如此就可以通过factory来与MyProtocolFactory交互
 def makeConnection(self,trans):
    print 'make connection: get transport: ', trans
    print 'my factory is: ', self.factory
     
 # 接口: 有数据到达
  def dataReceived(self, data):
    self.poem += data
    msg = 'Task %d: got %d bytes of poetry from %s'
    print msg % (self.task_num, len(data), self.transport.getPeer())
  
 # 接口: 连接断开
  def connectionLost(self, reason):
    # 连接断开的处理
 
 
class MyProtocolFactory(ClientFactory):
 
 # 接口: 通过protocol类成员指出需要创建的Protocols
  protocol = PoetryProtocol # tell base class what proto to build
 
  def __init__(self, address):
    self.poetry_count = poetry_count
    self.poems = {} # task num -> poem
     
 # 接口: 在创建Protocols的回调
  def buildProtocol(self, address):
    proto = ClientFactory.buildProtocol(self, address)
    # 在这里对proto做一些初始化....
    return proto
     
 # 接口: 连接Server失败时的回调
  def clientConnectionFailed(self, connector, reason):
    print 'Failed to connect to:', connector.getDestination()
     
def main(address):
 factory = MyClientFactory(address)
  host, port = address
  # 连接服务端时传入ProtocolsFactory
  reactor.connectTCP(host, port, factory) 
  reactor.run()

例 2 は、ネットワーク IO を処理する必要がなく、ロジックがより明確であるため、例 1 よりもはるかに単純かつ明確です。具体的には、ClientFactory と Protocol は、より柔軟で強力なロジック制御を実現するために、より多くのインターフェイスを提供します。インターフェイスは、ツイストされたソース コードにあります。

四. twisted Deferred
twisted Deferred对象用于解决这样的问题：有时候我们需要在ProtocolsFactory中嵌入自己的回调，以便Protocols中发生某个事件(如所有Protocols都处理完成)时，回调我们指定的函数(如TaskFinished)。如果我们自己来实现回调，需要处理几个问题:
如何区分回调的正确返回和错误返回?(我们在使用异步调用时，要尤其注意错误返回的重要性)
如果我们的正确返回和错误返回都需要执行一个公共函数(如关闭连接)呢?
如果保证该回调只被调用一次?
Deferred对象便用于解决这种问题，它提供两个回调链，分别对应于正确返回和错误返回，在正确返回或错误返回时，它会依次调用对应链中的函数，并且保证回调的唯一性。

d = Deferred()
# 添加正确回调和错误回调
d.addCallbacks(your_ok_callback, your_err_callback)
# 添加公共回调函数
d.addBoth(your_common_callback)
 
# 正确返回 将依次调用 your_ok_callback(Res) -> common_callback(Res)
d.callback(Res)
# 错误返回 将依次调用 your_err_callback(Err) -> common_callback(Err)
d.errback(Err)
 
# 注意，对同一个Defered对象，只能返回一次，尝试多次返回将会报错

twisted的defer是异步的一种变现方式，可以这么理解，他和thread的区别是，他是基于时间event的。
有了deferred，即可对任务的执行进行管理控制。防止程序的运行，由于等待某项任务的完成而陷入阻塞停滞，提高整体运行的效率。
Deferred能帮助你编写异步代码，但并不是为自动生成异步或无阻塞的代码！要想将一个同步函数编程异步函数，必须在函数中返回Deferred并正确注册回调。

五.综合示例

下面的例子，你们自己跑跑，我上面说的都是一些个零散的例子，大家对照下面完整的，走一遍。 twisted理解其实却是有点麻烦，大家只要知道他是基于事件的后，慢慢理解就行了。

#coding:utf-8
#xiaorui.cc
from twisted.internet import reactor, defer
from twisted.internet.threads import deferToThread
import os,sys
from twisted.python import threadable; threadable.init(1)
deferred =deferToThread.__get__
import time
def todoprint_(result):
  print result
def running():
  "Prints a few dots on stdout while the reactor is running."
#   sys.stdout.write("."); sys.stdout.flush()
  print '.'
  reactor.callLater(.1, running)
@deferred
def sleep(sec):
  "A blocking function magically converted in a non-blocking one."
  print 'start sleep %s'%sec
  time.sleep(sec)
  print '\nend sleep %s'%sec
  return "ok"
def test(n,m):
  print "fun test() is start"
  m=m
  vals = []
  keys = []
  for i in xrange(m):
    vals.append(i)
    keys.append('a%s'%i)
  d = None
  for i in xrange(n):
    d = dict(zip(keys, vals))
  print "fun test() is end"
  return d
if __name__== "__main__":
#one
  sleep(10).addBoth(todoprint_)
  reactor.callLater(.1, running)
  reactor.callLater(3, reactor.stop)
  print "go go !!!"
  reactor.run()
#two
  aa=time.time()
  de = defer.Deferred()
  de.addCallback(test)
  reactor.callInThread(de.callback,10000000,100 )
  print time.time()-aa
  print "我这里先做别的事情"
  print de
  print "go go end"

更多剖析Python的Twisted框架的核心特性相关文章请关注PHP中文网！