使用twisted编写异步服务器

￡神魔★判官ぃ 2022-08-21 05:59 146阅读 0赞

一 简介  
twisted是python下的一个事件驱动的异步网络应用框架，其项目主页为[http://twistedmatrix.com/trac/][http_twistedmatrix.com_trac]。 网络上关于twisted的介绍很多， 我这里就不啰嗦了。我们先来看一个使用twisted的实例。

# -*- coding: cp936 -*-
    import os, sys, time
    from twisted.internet.protocol import Protocol,Factory
    from twisted.internet import reactor, defer, threads
    from twisted.protocols.basic import LineReceiver
    ''' 网络连接建立后的，处理方法。该类原始继承为Protocol，当连接建立、断开、数据到达时会自动调用对应的处理函数 '''
    class EchoProtocol(LineReceiver):
    
        def connectionMade(self):
            self.sendLine("welcome")
        def lineReceived(self, line):
            self.sendLine(time.strftime("%H:%M:%S")+'server: '+line)
    ''' 一个网络服务只会实例化一个Factory，对于与连接无关的公用部分， 可放到该部分实现 '''
    class EchoFactory(Factory):
        def buildProtocol(self, addr):
            return EchoProtocol()
    if __name__ == '__main__':
        reactor.listenTCP(8007, EchoFactory()) 
        reactor.run() 
    else:
        from twisted.application import internet, service
        application = service.Application('echo')
        echoService = internet.TCPServer(8007, EchoFactory())
        echoService.setServiceParent(application)

在该例子中，我们可以使用python来启动该程序，也可以直接使用twisted来启动服务。如果使用twisted来启动服务，可以对是否作为守护进程、日志和pid文件的记录位置、事件驱动方式等项目进行配置， 可以使用twistd –help来查看具体可配置的栏目。  
二 使用thread  
前面我们提到，twisted采用的是纯事件驱动的一种模式，这意味着一个事件未处理完毕的情况下，是不能够在处理新事件的。但并不是所有事情都能马上或在短时间内做完，例如读写文件等。那么我们是否就无所适从了，答案是否定的。twisted的提供了defertothread函数来将阻塞式变为非阻塞的。  
(1)阻塞式  
\#在该函数退出之前，既不会写回数据，也不会处理新的连接

def lineReceived(self, line):
            self.sendLine('server: '+line)
            time.sleep(100000)

（2）非阻塞式  
\#在该函数会立刻结束，并在新的线程中执行time.sleep

def lineReceived(self, line):         
        self.sendLine('server: '+line)
        threads.deferToThread(time.sleep, 100000)

现在我们已经可以将阻塞式的函数变化为非阻塞式的了， 那么接下来的问题是如果阻塞函数执行完毕后需要给我返回结果怎么办？其实deferToThread对象返回的是一个defer， 我们可以添加callBack来完成该操作。  
三 使用defer  
　　　defer可以说是twisted里面最神奇的东西之一了。 使用defer对于， 即使数据并不立刻返回，但程序也不用等待，而是可以继续响应其他时间，待数据就绪后再处理数据。前面提到的defertothread函数其实返回的就是一个deferred对象。

def lineReceived(self, line):
        self.sendLine('server: '+line)
        d = mySleep(100000)
        d.addCallback(self.sendLine, 'Suc').addErrback(self.sendLine, 'Err')
    def mySleep(self, second):
        deferred = Deferred()
        threads.deferToThread(time.sleep, second)
        return deferred

deferred对象本身并不将函数变为非阻塞式函数，它只是记录了登记了一个事件位置，使得程序可以等待一定时间后在处理前置事件。

四 使用defer进行网络请求  
网络操作往往是需要消耗时间的，因为你并不知道网络何时建立、数据何时就绪。那么我们除了使用defertoThread来进行网络请求外，还能怎么进行网络请求呢。现在我们模拟这么一种场景：服务器接收到一个指令， 该指令为对用户登录进行校验， 如果校验通过则返回0, 否则-1。但用户的登录校验需要向另外一个服务器b发送命令来进行。  
方法一: 采用阻塞式网络请求

def lineReceived(self, line):
            import socket
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.connect( ('127.0.0.1', 8006) )
            sock.settimeout(60)
            sock.send(line+'\n')
            try:
                result = sock.recv(4096)
            except:
                result = None
            self.sendLine('0' if result else '-1')

方法二: 采用twisted connect请求

def lineReceived(self, line):
            reactor.connectTCP('127.0.0.1', 8006, VerifyFactory(line, self))   
    
    class VerifyProtocol(LineReceiver):
        def __init__(self, fatcory):
           self._factory = factory
        def connectionMade(self):
            self.sendLine(self._factory.line)
        def lineReceived(self, line):
            self._factory.request.sendLine('0' if line else '-1')
    
    class VerifyFactory(Factory):
        def __init__(self, line, requect):
            self.line = line
            self.request = request
        def buildProtocol(self, addr):
            return VerifyProtocol(self)

方法三: 采用deferred

from collections import deque
    class EchoProtocol(LineReceiver):
        def __init__(self, fatcory):
            self._factory = factory
        def connectionMade(self):
            self.sendLine("welcome")
        def lineReceived(self, line):
            self._factory.verify.verify(line).addCallback(self.verify_ret)
        def verify_ret(self, result):
            self.sendLine('0' if result else '-1')
    
    class EchoFactory(Factory):
        def __init__(self) :
            self.verfiy = None
            reactor.connectTCP('127.0.0.1', 8006, VerifyFactory(line, self))
        def buildProtocol(self, addr):
            return EchoProtocol(self)
    
    class VerifyProtocol(LineReceiver):
        def __init__(self, fatcory):
            self._factory = factory 
            self._current = deque()
        def connectionMade(self):
            self._factory.echo.verfiy = self
        def verify(self, user):
            deferred = Deferred()
            self._current.append(deferred)
             self.sendLine(user)
            return deferred
        def lineReceived(self, line):
            deferred = self._current.popleft()
           deferred.callback(line)
    
    class VerifyFactory(Factory):
        def __init__(self, echo):
            self.echo = echo
        def buildProtocol(self, addr):
            return VerifyProtocol(self)

方法三最关键的三个地方是self.\_current.append(deferred)、self.\_current.popleft()和self.\_factory.echo.verfiy = self ， 前两个是保证处理正确的deferred，后面一个是保证请求能够进行验证请求。

五 结束语  
　　　twisted是一个非常强大的网络框架，这里我们只是介绍了其最简单的应用，大家可以根据自己的需要研究相关的协议， 后面我们会对defer和基于defer的网络连接池做进一步详细的讲解。

[http_twistedmatrix.com_trac]: http://twistedmatrix.com/trac/