twisted下使用网络连接池

待我称王封你为后i 2022-08-21 06:36 115阅读 0赞

通过《使用twisted编写异步服务器》一文的学习，我们可以轻松使用twisted来搭建自己的服务器。前面的示例中服务端并不涉及到与外部服务器的交互，而实际的服务端往往需要同时与其他服务端进行交互（例如进行密码验证）。使用传统的os.socket连接服务器的方法虽然可行，但却并不符合异步的原则，使用deferred则可以异步的建立网络连接，频繁的网络连接总是会加大服务器压力，因此更多的时候是使用网络连接池。这里我们主要学习在twisted下如何异步的网络连接池。

class simpleClientFactory(protocol.ClientFactory):
        def __init__(self, simple_client_pool, d):
            self._deferred = d #传入一个deferr对象用于回调
            self.pool = simple_client_pool
        def buildProtocol(self, addr):
            return simpleClientProtocol(self) #将自身传递给protocol
        def clientConnectionFailed(self, connector, reason):
            self.pool.release(None)
            self._deferred.errback(reason)

与服务器端一个factory可对应多个protocol不同， client端一个factory只对应于一个protocol。我一般在protocol初始化的时候将factory对象传入，以便protocol可访问factory中的对象（protocol默认其实就绑定了factory，这里完全是个人习惯）。网络连接建立后，回调将protocol自身返回，调用者即可该protocol的sendLine发送命令到服务器。os.socket发送完命令后，调用者只需阻塞自己等待结果返回就可以，但sendLine会立刻返回，调用者并不知道自己什么时候可以得到结果也不知道什么时候可以拿到结果，这无疑是见郁闷的事情。我们可在sendLine上在封装一层sendCmd，sendCmd返回一个deferred对象，数据到达后回调该deferred即可。下面的示例实际在sendCmd上又封装了一次，这样客户端根据需要执行相应的命令即可。pool相关是为建立连接池时使用，这里暂时不用管。

class simpleClientProtocol(LineReceiver):
        delimiter = '\n'
        def __init__(self, factory):
            self._current = deque() # 用于存放sendCmd中返回的deferred对象， 以便数据返回时可以处理
            self._factory = factory
        def connectionMade(self):
            self._factory._deferred.callback(self)
        def connectionLost(self, reason):
            #如果该连接在连接池中，则从池中删除， 否则释放该连接，以唤醒等待队列
            try:
                self._factory.pool._pool.remove(self)
            except:
                self._factory.pool.release(None)
        def get(self):
            return self.sendCmd('get', 'get %s' % key)
        def sendCmd(self, cmd, fullcmd):
            cmdObj = Command(cmd, fullcmd)
            #防止在连串操作过程中连接丢失 
            if not self.transport.connected:
                cmdObj._deferred.errback(-1)
                return cmdObj._deferred
            self._current.append(cmdObj)
            self.sendLine(fullcmd)
            self.transport.doWrite()
            return cmdObj._deferred
        def lineReceived(self, data):
            cmd = self._current.popleft()
            cmd.success(data, self)
    class Command(object):
        def __init__(self):
            self._deferred = Deferred()
        def success(self, value, client): #这里可以对返回结果做判断，决定是callback还是errback
            self._deferred.callback(value)
        def err(self, error):
            self._deferred.errback(error)

一般连接池初始化应包括连接所需要的服务器信息、池的大小、连接最长空闲时间、连接最长时间等信息， 这里我们不考虑连接最长空间时间和连接最长时间。 从池中获取有效连接时有三种情况：1.连接池中存在有效链接时，从连接池获取；2.当连接数不超过最大连接数时候，尝试建立新连接；3.连接数达到最大，放入等待队列等候处理。用户在使用完该连接后需主动将连接归还给连接池，在归还连接时， 该连接有可能已经失效。连接池对象一般作为一个全局性的对象或者在需要地方将其传入。

class simple_client_pool_t(object):
        def __init__(self, host, port, capacity):
            self._host = host
            self._port = port
            self._capacity = capacity
            self._in_use = 0
            self._pool = []
            self._waitlist = deque()
        def create_simple_client(self, deferred):
            simple_client_factory = simpleClientFactory(self, deferred)
            reactor.connectTCP(self._host, self._port, simple_client_factory)
        def aquire(self, deferred=None): #从等待队列过来的会自行带deferred
            if not deferred:
                deferred = Deferred()
            if self._pool:
                simple_client = self._pool.pop()
                self._in_use += 1
                deferred.callback(simple_client)
            elif self._in_use < self._capacity:
                self._in_use += 1
                self.create_simple_client(deferred)
            else:
                self._waitlist.append(deferred)
            return deferred
        def release(self, simple_client):
            if not simple_client: #释放的client已经失效
                self._in_use -= 1
                if self._waitlist:
                    self.aquire(self._waitlist.popleft())
                return
            if self._waitlist: #等待队列不为空， 唤醒第一个等待
                self._waitlist.popleft().callback(simple_client)
            else:
                self._pool.append(simple_client)
                self._in_use -= 1

我们来看看如何从连接池中获取连接，并使用该连接进行网络交互。

def get(key):
        def acquire_suc(client, key):
             client.get(key).addCallback(get_suc, key).addErrback(acquire_err)    
        def acquire_err(result): 
            deferred.errback(result) #deferred对象自动访问到外部的deferred， 变量作用域可查看这里
    
        def get_suc(value):
            deferred.errback(value)
    
        deferred = Deferred()
        client_pool.aquire().addCallback(acquire_suc, key).addErrback(acquire_err) 
        return deferred
    
    client_pool = simple_client_pool_t('127.0.0.1', 8000, 10)
    get('key').addCallback(get_suc).addErrback(get_err) #这里未给出get_suc和get_err的实现， 但需将连接归还

通过前面的介绍，我们就可以在自己的twisted服务器中使用连接池与其他服务器进行交互，以减轻服务器压力，加快服务器响应速度。