我有一个基于Python的SimpleXMLRPCServer,类似于:
from multiprocessing import Process
from SimpleXMLRPCServer import SimpleXMLRPCServer
from SimpleXMLRPCServer import SimpleXMLRPCRequestHandler
import SocketServer
class RPCThreading(SocketServer.ThreadingMixIn, SimpleXMLRPCServer):
pass
# Restrict to a particular path.
class RequestHandler(SimpleXMLRPCRequestHandler):
rpc_paths = ('/RPC2',)
def main():
server = RPCThreading(('127.0.0.1', 8000), requestHandler=RequestHandler)
server.register_function(tester1)
server.register_function(tester2)
print("Server running...")
server.serve_forever()
def tester1(id):
p = Process(target=my_func1, args=(id,))
p.start()
return True
def tester2(id):
p = Process(target=my_func2, args=(id,))
p.start()
return True我想实现一种方法来跟踪当前有多少并发进程正在为tester1和tester2执行,如果它们中仍有超过最大(用户定义)数量的进程在执行,则对每个新请求进行排队,并在该数量低于阈值时执行。
也许每个函数都有一个共享的Pool?
发布于 2020-03-13 05:06:55
下面的代码似乎完成了我所要求的操作:
from multiprocessing import Process, JoinableQueue
from SimpleXMLRPCServer import SimpleXMLRPCServer
from SimpleXMLRPCServer import SimpleXMLRPCRequestHandler
import SocketServer
import threading
tester1_queue = JoinableQueue()
tester2_queue = JoinableQueue()
tester1_max_concurrent = 10
tester2_max_concurrent = 10
class RPCThreading(SocketServer.ThreadingMixIn, SimpleXMLRPCServer):
pass
# Restrict to a particular path.
class RequestHandler(SimpleXMLRPCRequestHandler):
rpc_paths = ('/RPC2',)
def main():
# spin up tester1 queue watcher threads
for i in range(tester1_max_concurrent):
worker = threading.Thread(target=tester1_queue_watcher, args=(tester1_queue,))
worker.daemon = True
worker.start()
# spin up tester2 queue watcher threads
for i in range(tester2_max_concurrent):
worker = threading.Thread(target=tester2_queue_watcher, args=(tester2_queue,))
worker.daemon = True
worker.start()
server = RPCThreading(('127.0.0.1', 8000), requestHandler=RequestHandler)
server.register_function(tester1_handler, 'tester1')
server.register_function(tester2_handler, 'tester2' )
print("Server running...")
server.serve_forever()
def tester1_handler(id):
tester1_queue.put((id,))
return True
def tester1_queue_watcher(q):
while True:
id = q.get()
p = Process(target=tester1, args=(id,))
p.start()
p.join()
q.task_done()
def tester1(id):
# do stuff
def tester2_handler(id):
tester2_queue.put((id,))
return True
def tester2_queue_watcher(q):
while True:
id = q.get()
p = Process(target=tester2, args=(id,))
p.start()
p.join()
q.task_done()
def tester2(id):
# do stuffhttps://stackoverflow.com/questions/60510856
复制相似问题