python - 如何在与主程序不同的线程中编写套接字服务器(使用gevent)？

Question

我正在开发Flask/gevent WSGIserver网络服务器，该服务器需要使用XML通过两个套接字与硬件设备(在后台)进行通信。

一个套接字由客户端(我的应用程序)启动，我可以将XML命令发送到设备。设备在另一个端口上应答，然后发回我的应用程序必须确认的信息。因此，我的应用程序必须监听第二个端口。

到目前为止，我已经发出命令，打开了第二个端口作为服务器，等待设备的响应并关闭了第二个端口。

问题在于设备有可能发送多个我必须确认的响应。因此，我的解决方案是保持端口开放并保持对传入请求的响应。但是，最终设备已完成发送请求，而我的应用程序仍在监听(我不知道设备何时完成)，从而阻止了其他所有内容。

这似乎是一个线程的完美用例，因此我的应用程序在单独的线程中启动了监听服务器。因为我已经将gevent用作Flask的WSGI服务器，所以可以使用greenlets。

问题是，我一直在寻找这样的事情的一个很好的例子，但是我所能找到的只是单个套接字服务器的多线程处理程序的例子。我不需要在套接字服务器上处理很多连接，但是我需要在单独的线程中启动它，以便它可以监听和处理传入的消息，而我的主程序可以继续发送消息。
我遇到的第二个问题是在服务器中，我需要使用“主”类中的某些方法。作为Python的新手，我不确定如何以使其成为可能的方式来构造它。

class Device(object):

def __init__(self, ...):
    self.clientsocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    self.serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

def _connect_to_device(self):
    print "OPEN CONNECTION TO DEVICE"
    try:
        self.clientsocket.connect((self.ip, 5100))
    except socket.error as e:
        pass

def _disconnect_from_device(self):
    print "CLOSE CONNECTION TO DEVICE"
    self.clientsocket.close()

def deviceaction1(self, ...):
    # the data that is sent is an XML document that depends on the parameters of this method.
    self._connect_to_device()
    self._send_data(XMLdoc)
    self._wait_for_response()
    return True

def _send_data(self, data):
    print "SEND:"
    print(data)
    self.clientsocket.send(data)

def _wait_for_response(self):
    print "WAITING FOR REQUESTS FROM DEVICE (CHANNEL 1)"
    self.serversocket.bind(('10.0.0.16', 5102))
    self.serversocket.listen(5)                         # listen for answer, maximum 5 connections
    connection, address = self.serversocket.accept()
    # the data is of a specific length I can calculate
    if len(data) > 0:
        self._process_response(data)
        self.serversocket.close()

def _process_response(self, data):
    print "RECEIVED:"
    print(data)
    # here is some code that processes the incoming data and
    # responds to the device
    # this may or may not result in more incoming data



if __name__ == '__main__':
    machine = Device(ip="10.0.0.240")
    Device.deviceaction1(...)

这就是(在全局范围内，我忽略了敏感信息)我现在正在做的事情。如您所见，一切都是顺序的。
如果任何人都可以在单独的线程中提供监听服务器的示例(最好使用greenlets)，以及从监听服务器返回到生成线程的通信方式，则将大有帮助。

谢谢。

编辑:
在尝试了几种方法之后，我决定使用Python的默认 select()方法来解决此问题。这行得通，所以我有关线程使用的问题不再重要。感谢为您的时间和精力提供投入的人们。

Answer 1

希望它能提供帮助。在示例类中，如果我们将调用tenMessageSender函数，则它将在不阻塞主循环的情况下启动异步线程，然后_zmqBasedListener将开始在单独的端口上监听，直到该线程处于 Activity 状态。以及tenMessageSender函数将发送的任何消息，这些消息将被客户端接收并响应zmqBasedListener。

服务器端

import threading
import zmq
import sys

class Example:
    def __init__(self):
        self.context = zmq.Context()
        self.publisher = self.context.socket(zmq.PUB)
        self.publisher.bind('tcp://127.0.0.1:9997')
        self.subscriber = self.context.socket(zmq.SUB)
        self.thread = threading.Thread(target=self._zmqBasedListener)

    def _zmqBasedListener(self):
        self.subscriber.connect('tcp://127.0.0.1:9998')
        self.subscriber.setsockopt(zmq.SUBSCRIBE, "some_key")
        while True:
            message = self.subscriber.recv()
            print message
        sys.exit()

    def tenMessageSender(self):
        self._decideListener()
        for message in range(10):
            self.publisher.send("testid : %d: I am a task" %message)

    def _decideListener(self):
        if not self.thread.is_alive():
            print "STARTING THREAD"
            self.thread.start()

客户

import zmq
context = zmq.Context()

subscriber = context.socket(zmq.SUB)
subscriber.connect('tcp://127.0.0.1:9997')
publisher = context.socket(zmq.PUB)
publisher.bind('tcp://127.0.0.1:9998')
subscriber.setsockopt(zmq.SUBSCRIBE, "testid")
count = 0
print "Listener"
while True:
    message = subscriber.recv()
    print message
    publisher.send('some_key : Message received %d' %count)
    count+=1

除了线程，您可以使用greenlet等。