带有线程池服务器python的套接字

Question

我有一个简单的多线程服务器，但它为每个套接字创建一个新线程，我不想创建很多线程。我的想法是以其他方式接收消息：当用户发送消息时，它会将消息添加到消息队列中，服务器将使用线程池处理这些请求。

简单的多线程服务器：

import socket
import threading

class ThreadedServer(object):
    def __init__(self, host, port):
        self.host = host
        self.port = port
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.sock.bind((self.host, self.port))

    def listen(self):
        self.sock.listen(5)
        while True:
            client, address = self.sock.accept()
            client.settimeout(60)
            threading.Thread(target = self.listenToClient,args = (client,address)).start()

    def listenToClient(self, client, address):
        size = 1024
        while True:
            try:
                data = client.recv(size)
                if data:
                    # Set the response to echo back the recieved data 
                    response = data
                    client.send(response)
                else:
                    raise error('Client disconnected')
            except:
                client.close()
                return False

if __name__ == "__main__":
    port_num = input("Port? ")
    ThreadedServer('',port_num).listen()

如何实现我的想法或有更好的方法来实现它？

Answer 1

这个问题似乎很老了，但我在处理套接字服务器时也偶然发现了同样的问题，所以这里是下面的代码，您可以使用它来制作线程套接字服务器，它在到达时不会产生新线程。

只是为了给出要点 ThreadingMixIn 类被线程池覆盖。

class ThreadPoolMixIn(socketserver.ThreadingMixIn):
'''
use a thread pool instead of a new thread on every request
'''
# numThreads = 50    
allow_reuse_address = True  # seems to fix socket.error on server restart
def serve_forever(self):
    '''
    Handle one request at a time until doomsday.
    '''
    print('[X] Server is Running with No of Threads :- {}'.format(self.numThreads))
    # set up the threadpool
    self.requests = Queue(self.numThreads)

    for x in range(self.numThreads):
        t = threading.Thread(target = self.process_request_thread)
        t.setDaemon(1)
        t.start()

    # server main loop
    while True:
        self.handle_request()
    self.server_close()

def process_request_thread(self):
    '''
    obtain request from queue instead of directly from server socket
    '''
    while True:
        socketserver.ThreadingMixIn.process_request_thread(self, *self.requests.get())

def handle_request(self):
    '''
    simply collect requests and put them on the queue for the workers.
    '''
    try:
        request, client_address = self.get_request()
    except socket.error:
        return
    if self.verify_request(request, client_address):
        self.requests.put((request, client_address))

然后在ThreadedTCPRequest Handler中调用并覆盖numThreads参数：

class ThreadedTCPServer(ThreadPoolMixIn, socketserver.TCPServer):
#Extend base class and overide the thread paramter to control the number of threads.
def __init__(self, no_of_threads, server_address, ThreadedTCPRequestHandler):
    self.numThreads = no_of_threads
    super().__init__(server_address, ThreadedTCPRequestHandler)

最终创建永远服务的服务器：

def create_multi_threaded_socket(CONFIG, HandlerClass = ThreadedTCPRequestHandler,
    ServerClass = ThreadedTCPServer, 
    protocol="HTTP/1.0"):

server_address = ('', CONFIG.port)
HandlerClass.protocol_version = protocol
# httpd = ServerClass(server_address, HandlerClass)
server = ThreadedTCPServer(CONFIG.no_of_threads, server_address, ThreadedTCPRequestHandler)
sa = server.socket.getsockname()
print("Serving HTTP on {} port : {}".format(sa[0], sa[1]))
server.serve_forever()

我的示例代码来自： http://code.activestate.com/recipes/574454-thread-pool-mixin-class-for-use-with-socketservert/

根据我的需要进行了修改。 希望这会有所帮助:)。