如何限制Twisted中的同时连接数

Question

所以我建立了一个扭曲的服务器,我想知道限制同时连接数的最佳方法是什么？

让我的工厂回归没有最好的方法吗？当我这样做时,我抛出了很多例外,例如:

exceptions.AttributeError: 'NoneType' object has no attribute 'makeConnection'

我想在某种程度上让客户只是坐在队列中,直到当前的连接数回落,但我不知道如何异步.

目前我正在使用我的工厂这样做:

class HandleClientFactory(Factory):

    def __init__(self):
            self.numConnections = 0 

    def buildProtocol(self, addr):
            #limit connection number here
            if self.numConnections >= Max_Clients:
                    logging.warning("Reached maximum Client connections")
                    return None

            return HandleClient(self)

哪个有效,但断开连接而不是等待,并且还会抛出许多未处理的错误.

Answer 1

你必须自己构建它.幸运的是,这些碎片大部分都是这样做的(你可能会要求稍微更合适的碎片但是......)

首先,要避免AttributeError(确实导致连接关闭),请务必IProtocol从您的buildProtocol方法返回提供程序.

class DoesNothing(Protocol):
    pass

class YourFactory(Factory):
    def buildProtocol(self, addr):
        if self.currentConnections < self.maxConnections:
            return Factory.buildProtocol(self, addr)
        protocol = DoesNothing()
        protocol.factory = self
        return protocol

如果您使用此工厂(填写缺失的部分 - 例如,初始化maxConnections并currentConnections正确跟踪),那么您将发现一旦达到限制就连接的客户端将获得DoesNothing协议.他们可以根据自己的喜好向此协议发送尽可能多的数据.它将丢弃所有.它永远不会向他们发送任何数据.它将使连接保持打开状态,直到它们关闭它.简而言之,它什么都不做.

但是,一旦连接数低于限制,您还希望客户实际接收服务.

要做到这一点,你需要更多的东西:

• 您必须保留可能发送缓冲的任何数据,以便在您准备好阅读时可以读取它们.
• 您必须跟踪连接,以便在时机成熟时开始为它们提供服务.
• 你必须在那个时候开始为他们服务.

对于第一个,您可以使用大多数传输的功能来"暂停":

class PauseTransport(Protocol):
    def makeConnection(self, transport):
        transport.pauseProducing()

class YourFactory(Factory):
    def buildProtocol(self, addr):
        if self.currentConnections < self.maxConnections:
            return Factory.buildProtocol(self, addr)
        protocol = PauseTransport()
        protocol.factory = self
        return protocol

PauseTransport类似于DoesNothing但是具有次要(和有用)差异,一旦它连接到传输它就告诉传输暂停.因此,不会从连接中读取任何数据,并且只要您准备好处理它,它们都将保持缓冲状态.

对于下一个要求,存在许多可能的解决方案.最简单的方法之一是使用工厂作为存储:

class PauseAndStoreTransport(Protocol):
    def makeConnection(self, transport):
        transport.pauseProducing()
        self.factory.addPausedTransport(transport)

class YourFactory(Factory):
    def buildProtocol(self, addr):
        # As above
        ...

    def addPausedTransport(self, transport):
        self.transports.append(transport)

再次,通过正确的设置(例如,初始化transports属性),您现在拥有一个列表,其中包含与您在等待服务的并发限制之上接受的连接相对应的所有传输.

对于最后一个要求,所有必要的是实例化和初始化实际上能够为您的客户提供服务的协议.实例化很容易(这是你的协议,你可能知道它是如何工作的).初始化主要是调用makeConnection方法:

class YourFactory(Factory):
    def buildProtocol(self, addr):
        # As above
        ...
    def addPausedTransport(self, transport):
        # As above
        ...
    def oneConnectionDisconnected(self)
        self.currentConnections -= 1
        if self.currentConnections < self.maxConnections:
            transport = self.transports.pop(0)
            protocol = self.buildProtocol(address)
            protocol.makeConnection(transport)
            transport.resumeProducing()

我已经省略了跟踪address所需参数的详细信息buildProtocol(transport从它的起始点到程序的这一部分,如果你的程序实际上需要它,应该清楚如何为原始地址值做类似的事情).

除此之外,这里发生的一切都是你采取下一个排队的传输(你可以使用不同的调度算法,例如LIFO)并将其连接到你选择的协议,就像Twisted一样.最后,撤消先前的暂停操作,以便数据开始流动.

或者......差不多.除非Twisted传输实际上没有任何方式来改变它们传递数据的协议,否则这将非常灵活.因此,如所写的,来自客户端的数据实际上将被传递到原始PauseAndStoreTransport协议实例.你可以解决这个问题("hack"显然是正确的词).将传输和 PauseAndStoreTransport实例存储在工厂的列表中,然后:

    def oneConnectionDisconnected(self)
        self.currentConnections -= 1
        if self.currentConnections < self.maxConnections:
            originalProtocol, transport = self.transports.pop(0)
            newProtocol = self.buildProtocol(address)

            originalProtocol.dataReceived = newProtocol.dataReceived
            originalProtocol.connectionLost = newProtocol.connectionLost

            newProtocol.makeConnection(transport)
            transport.resumeProducing()

现在,传输要调用方法的对象已将其方法替换为您希望调用方法的对象的方法.再次,这显然是一个黑客.如果你愿意的话,你可以把不太讨厌的东西放在一起(例如,明确支持委托给另一个协议的第三个协议类).这个想法是一样的 - 它只会在你的键盘上磨损.对于它的价值,我怀疑使用Tubes做类似的事情可能更容易,更少打字,但我现在将尝试基于该库的解决方案.

我已经避免解决保持currentConnections适当更新的问题.既然你已经numConnections在你的问题中,我假设你知道如何管理那部分.我在这里的最后一步所做的就是假设您执行减量步骤的方式是调用oneConnectionDisconnected工厂.

我还避免解决排队连接变得无聊并消失的事件.这将主要按写入方式工作 - 在您调用之前,Twisted不会注意到连接已关闭resumeProducing,然后connectionLost将在您的应用程序协议上调用.这应该没问题,因为您的协议无论如何都需要处理丢失的连接.