如何限制Twisted中的同时连接数 [英] How to limit the number of simultaneous connections in Twisted

问题描述

所以我有一台扭曲的服务器，我想知道限制同时连接数的最佳方法是什么?

so I have a twisted server I built, and I was wondering what is the best way to limit the number of simultaneous connections?

让我的工厂退货是最好的办法吗?当我这样做时，我抛出了很多异常，例如:

Is having my Factory return None the best way? When I do this, I throw a lot of exceptions like:

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

我想以某种方式让客户端排在队列中，直到当前的连接数回落为止，但是我不知道如何异步进行.

I would like someway to have the clients just sit in queue until the current connection number goes back down, but I don't know how to do that asynchronously.

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

Currently I am using my factory do like this:

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)

可以工作，但会断开连接而不是等待，并且还会引发许多未处理的错误.

which works, but disconnects rather than waits, and also throws a lot of unhandled errors.

推荐答案

您必须自己构建它.幸运的是，这些零件大部分都已准备就绪(您可能会要求稍微合适些的零件，但是...)

You have to build this yourself. Fortunately, the pieces are mostly in place to do so (you could probably ask for slightly more suitable pieces but ...)

首先，为避免AttributeError(这确实导致连接关闭)，请确保从buildProtocol方法返回IProtocol提供程序.

First, to avoid the AttributeError (which indeed causes the connection to be closed), be sure to return an IProtocol provider from your buildProtocol method.

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协议.他们可以根据需要向该协议发送尽可能多的数据.它将全部丢弃.它永远不会向他们发送任何数据.它将使连接保持打开状态，直到他们将其关闭.简而言之，它什么都不做.

If you use this factory (filling in the missing pieces - eg, initializing maxConnections and so tracking currentConnections correctly) then you'll find that clients which connect once the limit has been reached are given the DoesNothing protocol. They can send as much data as they like to this protocol. It will discard it all. It will never send them any data. It will leave the connection open until they close it. In short, it does nothing.

但是，您还希望客户端一旦连接数低于限制就可以实际接收服务.

However, you also wanted clients to actually receive service once connection count fell below the limit.

要执行此操作，您还需要几块:

To do this, you need a few more pieces:

• 您必须将它们可能发送的所有数据保留在缓冲中，以便在准备读取时可以读取它们.
• 您必须跟踪连接，以便在时机成熟时开始为它们提供服务.
• 您必须在上述时间开始为他们提供服务.

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

For the first of these, you can use the feature of most transports to "pause":

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相似，但略有不同(且非常有用)的区别在于，一旦将其连接到运输工具，它便会告知运输工具暂停.因此，将永远不会从连接中读取任何数据，并且只要您准备好处理这些数据，它们都将保留在缓冲区中.

PauseTransport is similar to DoesNothing but with the minor (and useful) difference that as soon as it is connected to a transport it tells the transport to pause. Thus, no data will ever be read from the connection and it will all remain buffered for whenever you're ready to deal with it.

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

For the next requirement, many possible solutions exist. One of the simplest is to use the factory as storage:

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属性)，您现在有了所有传输的列表，这些列表与您在并发限制之上接受的等待服务的连接相对应.

Again, with the proper setup (eg, initialize the transports attribute), you now have a list of all of the transports which correspond to connections you've accepted above the concurrency limit which are waiting for service.

对于最后一个要求，所需要做的就是实例化和初始化实际上可以为您的客户提供服务的协议.实例化很容易(这是您的协议，您可能知道它是如何工作的).初始化很大程度上是调用makeConnection方法的问题:

For the last requirement, all that is necessary is to instantiate and initialize the protocol that's actually capable of serving your clients. Instantiation is easy (it's your protocol, you probably know how it works). Initialization is largely a matter of calling the makeConnection method:

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

我已经省略了跟踪buildProtocol所需的address自变量的详细信息(transport从其原始点传送到程序的这一部分，应该清楚如何做某事如果您的程序确实需要原始地址值，则类似.

I've omitted the details of keeping track of the address argument required by buildProtocol (with the transport carried from its point of origin to this part of the program, it should be clear how to do something similar for the original address value if your program actually wants it).

除此之外，这里发生的所有事情都是您进行下一个排队的传输(如果需要，可以使用其他调度算法，例如LIFO)，并将其连接到您选择的协议，就像Twisted一样.最后，您撤消之前的暂停操作，以便数据开始流动.

Apart from that, all that happens here is you take the next queued transport (you could use a different scheduling algorithm if you want, eg LIFO) and hook it up to a protocol of your choosing just as Twisted would do. Finally, you undo the earlier pause operation so data will begin to flow.

或者...差不多.除了Twisted传输实际上并没有公开任何方式来更改向其传送数据的协议之外，这将是非常巧妙的.因此，按照书面规定，来自客户端的数据实际上将传递到原始的PauseAndStoreTransport协议实例.您可以解决这个问题("hack"显然是正确的词).将运输和 PauseAndStoreTransport实例都存储在工厂的列表中，然后:

Or... almost. This would be pretty slick except Twisted transports don't actually expose any way to change which protocol they deliver data to. Thus, as written, data from clients will actually be delivered to the original PauseAndStoreTransport protocol instance. You can hack around this (and "hack" is clearly the right word). Store both the transport and PauseAndStoreTransport instance in the list on the factory and then:

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

现在，传输要调用方法的对象已将其方法替换为您要调用方法的对象中的方法.同样，这显然是一个hack.如果需要的话，您可能可以放些别具一格的东西(例如，第三个协议类，明确支持委派给另一个协议).想法是一样的-只会增加键盘的磨损.对于它的价值，我怀疑使用 Tubes 来执行类似的操作可能更容易，而且打字更少.现在，我将尝试将基于该库的解决方案交给其他人.

Now the object that the transport wants to call methods on has had its methods replaced by those from the object that you want the methods called on. Again, this is clearly a hack. You can probably put together something less hackish if you want (eg, a third protocol class that explicitly supports delegating to another protocol). The idea will be the same - it'll just be more wear on your keyboard. For what it's worth, I suspect that it may be both easier and less typing to do something similar using Tubes but I'll leave an attempt at a solution based on that library to someone else for now.

我已避免解决使currentConnections保持最新状态的问题.由于您的问题中已经有numConnections，因此我假设您知道如何管理该部分.我在这里最后一步中所做的一切都是假设您执行减量步骤的方法是在工厂中调用oneConnectionDisconnected.

I've avoided addressing the problem of keeping currentConnections properly up to date. Since you already had numConnections in your question I'm assuming you know how to manage that part. All I've done in the last step here is suppose that the way you do the decrement step is by calling oneConnectionDisconnected on the factory.

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

I've also avoided addressing the event that a queued connection gets bored and goes away. This will mostly work as written - Twisted won't notice the connection was closed until you call resumeProducing and then connectionLost will be called on your application protocol. This should be fine since your protocol needs to handle lost connections anyway.

这篇关于如何限制Twisted中的同时连接数的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！