作为客户端/服务器的问题的Twisted

3 投票
1 回答
1096 浏览
提问于 2025-04-16 17:03

我正在使用twisted来创建一个“客户端/服务器”程序,这个程序既可以作为客户端获取一些数据,也可以作为服务器重新发送这些数据或者发送其他数据给客户端。

我使用的是twistd,我的代码初始化工厂的部分看起来是这样的:

application = service.Application('monitorD', uid=0, gid=0)
factoryMonitord = ServerFactory(p)
internet.TCPServer(9000, factoryMonitord).setServiceParent(service.IServiceCollection(application))
#because I need to send some datas from client factory to clients of serverfactory
factoryScamd = ClientFactory(factoryMonitord)
internet.TCPClient("localhost", 8001, factoryScamd).setServiceParent(service.IServiceCollection(application))

我的问题是,当客户端无法连接(因为服务器不可用)时,似乎会“阻塞”我所有的服务器部分。虽然我的服务器部分和其他客户端之间仍然可以通信,但速度非常慢……(而当我的客户端能够连接时,一切都正常工作)。

谢谢大家的帮助。

编辑:

这是我的ServerFactory代码(里面有很多没用的代码……):

class ServerFactory(protocol.ServerFactory):
    protocol = ServerProtocol

    def __init__(self, portal):
        #self.tp = ClientFactory.tp 
        self.tp = []
        self.portal = portal
        self.loop_vol_raid_areca = LoopingCall(self.checkVolRaidAreca)
        self.loop_vol_raid_areca.start(30)
        self.loop_services = LoopingCall(self.checkServices)
        self.loop_services.start(30)

    def buildProtocol(self, addr):
        p = protocol.ServerFactory.buildProtocol(self, addr)
        p.portal = self.portal
        return p

    def sendList(self, data):
        if data:
            if isinstance(data, list):
                for element in data:
                    if isinstance(element, list):
                        self.clean_data = "".join(element)
                self.sendToClients(self.clean_data)

    def sendToClients(self, data):
        print "SEND to MonitorC from MonitorD, tp:", self.tp
        if data:
            for tp in self.tp:
                self.protocol.sendLine(tp, data)

    def checkServices(self):
        self.scamd = threads.deferToThread(checkScamd)
        self.scamd.addCallback(self.sendToClients)
        self.httpd = threads.deferToThread(checkHttpd)
        self.httpd.addCallback(self.sendToClients)
        self.postgres = threads.deferToThread(checkPostgres)
        self.postgres.addCallback(self.sendToClients)

    def checkVolRaidAreca(self):
        self.vol = threads.deferToThread(check_vol_areca)
        self.vol.addCallback(self.sendList)
        self.event = threads.deferToThread(get_last_event_areca)
        self.event.addCallback(self.sendList)

这是客户端工厂的代码,也有很多没用的代码:

class ClientFactory(protocol.ClientFactory):
    protocol = ClientProtocol

    def __init__(self, MonitordFactory):
        self.tp = MonitordFactory.tp

    def clientConnectionFailed(self, connector, reason):
        print "Connection to scamd failed - retrying..."
        time.sleep(30)
        connector.connect()

    def clientConnectionLost(self, connector, reason):
        print "Connection to scamd lost - retrying..."
        time.sleep(30)
        connector.connect()

    def getCamList(self, data):
        cams_list = data.split("\x00")
        self.cams_array = []
        for i in cams_list:
            if str.find(i, "camera") != -1:
                i = i.split(" ")
                i = filter(lambda x: len(x)>0, i)
                self.cams_array.append(i)

    def checkCams(self, data):
        data = data.split(" ")
        for i in self.cams_array:
            if i[1] == data[2]:
                if data[3] == "-1":
                    msg = i[6] + " ERREUR -1"
                if data[3] == "0":
                    msg = i[6] + " ERREUR 0"
                if data[3] == "1":
                    msg = i[6] + " ERREUR 1"
                if data[3] == "2":
                    msg = i[6] + " ERREUR 2 (RECO)"
        return msg

如果需要更多信息,我会把完整的代码发到pastebin上。还有,我是python和twisted的新手(但我以前用C或C++编程过)。

1 个回答

2

你可以立刻做两件事情来改善情况。首先,去掉 time.sleep(30) 这个调用。其次,你现在使用了很多线程池,所以你可能需要增加线程池的大小。每30秒你会创建5个线程,而默认的线程池大小也是5,这样你创建的其他线程就得在这5个线程后面排队等着。这可能是导致你的服务运行得比较慢,甚至看起来像是卡住的原因。

短期内,你可以在你的 ServerFactory 中创建一个自己的 twisted.python.threadpool.Threadpool

class ServerFactory(protocol.ServerFactory):
    protocol = ServerProtocol

    def __init__(self, portal):
        # ...
        self.threadpool = ThreadPool(25, 50)

    def checkServices(self):
        # ...
        self.postgres = threads.deferToThreadPool(reactor, self.threadpool, checkPostgres)

不过,从长远来看,尽量使用异步API来处理你的服务会更好。

比如,你的 checkHttpd 代码可以通过使用 twisted.web.client.Agent 来实现异步化。对于PostgreSQL,你可以使用 txPostgres。我对你提到的 scamd 服务不太熟悉,所以用异步IO连接它可能会比较复杂(也就是说,你可能需要为这个服务编写一个异步版本的协议)。

撰写回答