【问题标题】:Implement Per-Function Maximum Number of Concurrent Calls to Python XML-RPC Server实现对 Python XML-RPC 服务器的每个函数的最大并发调用数
【发布时间】:2020-06-16 01:09:33
【问题描述】:

我有一个基于 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?

【问题讨论】:

    标签: python multithreading python-multithreading xml-rpc simplexmlrpcserver


    【解决方案1】:

    以下似乎符合我的要求:

    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 stuff
    

    【讨论】:

      猜你喜欢
      • 2012-06-10
      • 1970-01-01
      • 2010-12-07
      • 1970-01-01
      • 2012-07-27
      • 2016-11-21
      • 1970-01-01
      • 2012-01-14
      • 1970-01-01
      相关资源
      最近更新 更多