如何正确触发python扭曲传输？

Question

我被要求编写一个连接到服务器的类，异步地向服务器发送各种命令，然后将返回的数据提供给客户机。我被要求用Python来做这个，Python对我来说是一种新的语言。我开始四处挖掘，发现了Twisted框架，它提供了一些非常好的抽象（协议、协议工厂、Reactor），它可以做很多我必须做的事情，如果我要推出自己的基于socket的应用程序的话。考虑到我要解决的问题，这似乎是正确的选择。在

我浏览了web上的许多例子（主要是Krondo），但是我仍然没有看到一个很好的例子来创建一个客户机，它将通过网络发送多个命令，并维护我创建的连接。在本例中，服务器（我无法控制）在发送响应后不会断开连接。那么，设计客户端的正确方法是什么呢？这样我就可以用各种方式逗服务器了？在

现在我要做的是：

class TestProtocol(Protocol)
    def connectionMade(self):
         self.transport.write(self.factory.message)

class TestProtocolFactory(Factory):
    message = ''
    def setMessage(self, msg):
        self.message = msg

def main():
    f = TestProtocolFactory()
    f.setMessage("my message")
    reactor.connectTCP(...)
    reactor.run()

{protocol{1）真正要调用的是另一个线程的连接，而不是调用。在

Answer 1

twisted框架是基于事件的编程，其方法本质上都是异步调用的，结果是由defer对象获得的。在

该框架的性质适合于协议开发，只需改变传统的顺序编程思想即可。Protocol类类似于一个有限状态机，其事件包括：connectionmake、connectionlost、receivedata。 然后将协议转换成适合你的客户端代码。在

下面是我想表达的一个粗略的例子。一点胭脂，但我现在可以提供：

class SyncTransport(Protocol):
    # protocol
    def dataReceived(self, data):
        print 'receive data', data
    def connectionMade(self):
        print 'i made a sync connection, wow'
        self.transport.write('x')
        self.state = I_AM_LIVING
    def connectionLost(self):
        print 'i lost my sync connection, sight'
    def send(self, data):
        if self.state == I_AM_LIVING:
            if data == 'x':
              self.transport.write('y')
           if data == 'Y':
              self.transport.write('z')
              self.state = WAITING_DEAD
        if self.state == WAITING_DEAD:
              self.transport.close()

Answer 2

您可能需要使用Service。在

服务是Twisted应用程序中启动和停止的部分功能，是代码的其他部分进行交互的很好的抽象。例如，在本例中，您可能有一个SayStuffToServerService（我知道，很糟糕的名称，但在不了解其工作的情况下，我只能在这里做到：））暴露了如下内容：

class SayStuffToServerService:
    def __init__(self, host, port):
        # this is the host and port to connect to

    def sendToServer(self, whatToSend):
        # send some line to the remote server

    def startService(self):
        # call me before using the service. starts outgoing connection efforts.

    def stopService(self):
        # clean reactor shutdowns should call this method. stops outgoing
        # connection efforts.

（这可能是您需要的所有接口，但应该非常清楚可以在哪里添加内容。）

这里的startService()和stopService()方法正是Twisted的服务公开的。另外，还有一个预先准备好的Twisted服务，它就像一个TCP客户机，为您处理所有的反应器。它是twisted.application.internet.TCPClient，它接受远程主机和端口的参数，以及处理实际连接尝试的协议工厂。在

以下是SayStuffToServerService，它作为TCPClient的子类实现：

^{2}$

（请参阅下面的SayStuffToServer协议工厂。）

使用这种服务架构在很多方面都很方便；您可以将服务组合在一个容器中，这样当您的应用程序的不同部分需要活动时，它们都可以作为一个容器停止和启动。将应用程序的其他部分作为单独的服务来实现可能很有意义。您可以将服务作为子服务设置为application-这是twistd查找的魔法名称，以便知道如何初始化、守护和关闭应用程序。实际上是的，现在我们添加一些代码来实现这个目的。在

from twisted.application import service

...

application = service.Application('say-stuff')

sttss = SayStuffToServerService('localhost', 65432)
sttss.setServiceParent(service.IServiceCollection(application))

仅此而已。现在，当您在twistd下运行这个模块（即，为了调试，twistd -noy saystuff.py），这个{}将在正确的reactor下启动，它将依次启动sayStuffToServer服务，它将启动一个连接工作来本地主机：65432，它将使用服务的factory属性来设置连接和协议。你不需要再打电话给reactor.run()或者自己把东西附在反应堆上了。在

所以我们还没有实现SayStuffToServerProtocolFactory。因为听起来您希望您的客户机在失去连接时重新连接（这样sendToServer的调用方通常可以假设有一个正常的连接），所以我将把这个协议工厂放在ReconnectingClientFactory之上。在

from twisted.internet import protocol

class SayStuffToServerProtocolFactory(protocol.ReconnectingClientFactory):
    _my_live_proto = None
    protocol = SayStuffToServerProtocol

这是一个非常好的最小定义，它将继续尝试与我们指定的主机和端口建立传出的TCP连接，并每次实例化一个saysufftotoserverprotocol。当我们连接失败时，这个类将执行良好的、良好的指数回退，这样您的网络就不会受到冲击（您可以设置最大等待时间）。协议将负责分配给_my_live_proto并调用此工厂的resetDelay()方法，以便指数回退将继续按预期工作。现在的协议是：

class SayStuffToServerProtocol(basic.LineReceiver):
    def connectionMade(self):
        # if there are things you need to do on connecting to ensure the
        # connection is "all right" (maybe authenticate?) then do that
        # before calling:
        self.factory.resetDelay()
        self.factory._my_live_proto = self

    def connectionLost(self, reason):
        self.factory._my_live_proto = None
        del self.factory

    def sayStuff(self, stuff):
        self.sendLine(stuff)

    def lineReceived(self, line):
        # do whatever you want to do with incoming lines. often it makes sense
        # to have a queue of Deferreds on a protocol instance like this, and
        # each incoming response gets sent to the next queued Deferred (which
        # may have been pushed on the queue after sending some outgoing
        # message in sayStuff(), or whatever).
        pass

这是在twisted.protocols.basic.LineReceiver之上实现的，但是如果您的协议不是面向行的，那么它也可以与任何其他类型的协议一起工作。在

唯一剩下的就是将服务连接到右侧的协议实例。这就是为什么工厂保留一个_my_live_proto属性，当连接成功建立时应该设置该属性，当连接丢失时清除（设置为None）。以下是SayStuffToServerService.sendToServer的新实现：

class NotConnectedError(Exception):
    pass

class SayStuffToServerService(internet.TCPClient):

    ...

    def sendToServer(self, whatToSend):
        if self.factory._my_live_proto is None:
            # define here whatever behavior is appropriate when there is no
            # current connection (in case the client can't connect or
            # reconnect)
            raise NotConnectedError
        self.factory._my_live_proto.sayStuff(whatToSend)

现在要把它们放在一个地方：

from twisted.application import internet, service
from twisted.internet import protocol
from twisted.protocols import basic

class SayStuffToServerProtocol(basic.LineReceiver):
    def connectionMade(self):
        # if there are things you need to do on connecting to ensure the
        # connection is "all right" (maybe authenticate?) then do that
        # before calling:
        self.factory.resetDelay()
        self.factory._my_live_proto = self

    def connectionLost(self, reason):
        self.factory._my_live_proto = None
        del self.factory

    def sayStuff(self, stuff):
        self.sendLine(stuff)

    def lineReceived(self, line):
        # do whatever you want to do with incoming lines. often it makes sense
        # to have a queue of Deferreds on a protocol instance like this, and
        # each incoming response gets sent to the next queued Deferred (which
        # may have been pushed on the queue after sending some outgoing
        # message in sayStuff(), or whatever).
        pass

class SayStuffToServerProtocolFactory(protocol.ReconnectingClientFactory):
    _my_live_proto = None
    protocol = SayStuffToServerProtocol

class NotConnectedError(Exception):
    pass

class SayStuffToServerService(internet.TCPClient):
    factoryclass = SayStuffToServerProtocolFactory

    def __init__(self, host, port):
        self.factory = self.factoryclass()
        internet.TCPClient.__init__(self, host, port, self.factory)

    def sendToServer(self, whatToSend):
        if self.factory._my_live_proto is None:
            # define here whatever behavior is appropriate when there is no
            # current connection (in case the client can't connect or
            # reconnect)
            raise NotConnectedError
        self.factory._my_live_proto.sayStuff(whatToSend)

application = service.Application('say-stuff')

sttss = SayStuffToServerService('localhost', 65432)
sttss.setServiceParent(service.IServiceCollection(application))

希望这能提供足够的框架来开始。有时需要做很多工作来处理客户机断开连接，或者处理来自服务器的无序响应，或者处理各种超时、取消挂起的请求、允许多个池连接等，但这应该会有所帮助。在

Answer 3

视情况而定。以下是一些可能性：

我假设

方法1。你有一个发送服务器的命令列表，但由于某些原因不能同时执行这些命令。在这种情况下，发送一个新的答案，作为前一个答案返回：

class proto(parentProtocol):
    def stringReceived(self, data):
        self.handle_server_response(data)
        next_command = self.command_queue.pop()
        # do stuff

方法2。发送到服务器的内容取决于服务器向您发送的内容：

^{2}$

方法3。服务器不会定期发送一些命令给您：

class proto(parentProtocol):
    def callback(self):
        next_command = self.command_queue.pop()
        # do stuff
    def connectionMade(self):
        from twisted.internet import task
        self.task_id = task.LoopingCall(self.callback)
        self.task_id.start(1.0)

方法4：你的编辑现在提到从另一个线程触发。请随意查看twisted文档，以确定proto.sendString是否是线程安全的。你可以直接打电话，但我不知道。方法3是线程安全的。只需从另一个线程填充队列（这是线程安全的）。在

基本上你可以在你的协议中存储任何数量的状态；它会一直存在直到你完成。你要么向服务器发送命令作为对它给你的消息的响应，要么你设置一些时间表来完成你的工作。或者两者兼而有之。在