如何在与主程序不同的线程中编写套接字服务器(使用gevent)? [英] How can I write a socket server in a different thread from my main program (using gevent)?

问题描述

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

I'm developing a Flask/gevent WSGIserver webserver that needs to communicate (in the background) with a hardware device over two sockets using XML.

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

One socket is initiated by the client (my application) and I can send XML commands to the device. The device answers on a different port and sends back information that my application has to confirm. So my application has to listen to this second port.

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

Up until now I have issued a command, opened the second port as a server, waited for a response from the device and closed the second port.

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

The problem is that it's possible that the device sends multiple responses that I have to confirm. So my solution was to keep the port open and keep responding to incoming requests. However, in the end the device is done sending requests, and my application is still listening (I don't know when the device is done), thereby blocking everything else.

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

This seemed like a perfect use case for a thread, so that my application launches a listening server in a separate thread. Because I'm already using gevent as a WSGI server for Flask, I can use the greenlets.

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

The problem is, I have looked for a good example of such a thing, but all I can find is examples of multi-threading handlers for a single socket server. I don't need to handle a lot of connections on the socket server, but I need it launched in a separate thread so it can listen for and handle incoming messages while my main program can keep sending messages. The second problem I'm running into is that in the server, I need to use some methods from my "main" class. Being relatively new to Python I'm unsure how to structure it in a way to make that possible.

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)，并提供一种从侦听服务器与派生线程进行通信的方法，那么它将大有帮助.

This is (globally, I left out sensitive information) what I'm doing now. As you can see everything is sequential. If anyone can provide an example of a listening server in a separate thread (preferably using greenlets) and a way to communicate from the listening server back to the spawning thread, it would be of great help.

谢谢.

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

After trying several methods, I decided to use Pythons default select() method to solve this problem. This worked, so my question regarding the use of threads is no longer relevant. Thanks for the people who provided input for your time and effort.

推荐答案

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

Hope it can provide some help, In example class if we will call tenMessageSender function then it will fire up an async thread without blocking main loop and then _zmqBasedListener will start listening on separate port untill that thread is alive. and whatever message our tenMessageSender function will send, those will be received by client and respond back to 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等代替线程.

Instead of thread you can use greenlet etc.

这篇关于如何在与主程序不同的线程中编写套接字服务器(使用gevent)?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！