我需要通过套接字接收连接,读取输入数据,进行长时间的艰苦计算,然后发送答案。同时进行的查询可能很多(即100条)
我知道,由于GIL,我不能使用普通线程,而是尝试将C++与boost:threads和boost:python一起使用,并在每个线程中运行python的子解释器。但是无论如何,它并不能同时100%使用所有内核。
因此,我决定使用多重处理,但是创建了一个静态的工作人员计数池,以通过队列为这些请求提供服务。这样,我们就不会浪费时间派生一个进程,并且我们不会同时有100个或更多的进程,而只有静态计数。
我是Python的新手,主要是我使用C++
所以现在我有了这段代码,但是它不起作用。连接打开并立即关闭,我不知道为什么:
#!/usr/bin/env python
import os
import sys
import SocketServer
import Queue
import time
import socket
import multiprocessing
from multiprocessing.reduction import reduce_handle
from multiprocessing.reduction import rebuild_handle
class MultiprocessWorker(multiprocessing.Process):
def __init__(self, sq):
self.SLEEP_INTERVAL = 1
# base class initialization
multiprocessing.Process.__init__(self)
# job management stuff
self.socket_queue = sq
self.kill_received = False
def run(self):
while not self.kill_received:
try:
h = self.socket_queue.get_nowait()
fd=rebuild_handle(h)
client_socket=socket.fromfd(fd,socket.AF_INET,socket.SOCK_STREAM)
#client_socket.send("hellofromtheworkerprocess\r\n")
received = client_socket.recv(1024)
print "Recieved on client: ",received
client_socket.close()
except Queue.Empty:
pass
#Dummy timer
time.sleep(self.SLEEP_INTERVAL)
class MyTCPHandler(SocketServer.BaseRequestHandler):
"""
The RequestHandler class for our server.
It is instantiated once per connection to the server, and must
override the handle() method to implement communication to the
client.
"""
def handle(self):
# self.request is the TCP socket connected to the client
#self.data = self.request.recv(1024).strip()
#print "{} wrote:".format(self.client_address[0])
#print self.data
# just send back the same data, but upper-cased
#self.request.sendall(self.data.upper())
#Either pipe it to worker directly like this
#pipe_to_worker.send(h) #instanceofmultiprocessing.Pipe
#or use a Queue :)
h = reduce_handle(self.request.fileno())
socket_queue.put(h)
if __name__ == "__main__":
#Mainprocess
address = ('localhost', 8082)
server = SocketServer.TCPServer(address, MyTCPHandler)
socket_queue = multiprocessing.Queue()
for i in range(5):
worker = MultiprocessWorker(socket_queue)
worker.start()
try:
server.serve_forever()
except KeyboardInterrupt:
sys.exit(0)
最佳答案
有没有不使用的理由
def reduce_socket(s):
...
def rebuild_socket(ds):
...
?
看来您可以这样做:
import copyreg
copyreg.pickle(type(socket.socket), reduce_socket, rebuild_socket)
然后将套接字传递给队列。
这些是建议。他们有帮助吗?
关于python - Python套接字多处理程序工作池,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/23107854/