首页
学习
活动
专区
圈层
工具
发布
社区首页 >问答首页 >实现每个函数对Python XML-RPC服务器的最大并发调用数

实现每个函数对Python XML-RPC服务器的最大并发调用数
EN

Stack Overflow用户
提问于 2020-03-03 23:46:13
回答 1查看 62关注 0票数 0

我有一个基于Python的SimpleXMLRPCServer,类似于:

代码语言:javascript
复制
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

我想实现一种方法来跟踪当前有多少并发进程正在为tester1tester2执行,如果它们中仍有超过最大(用户定义)数量的进程在执行,则对每个新请求进行排队,并在该数量低于阈值时执行。

也许每个函数都有一个共享的Pool

EN

Stack Overflow用户

回答已采纳

发布于 2020-03-13 05:06:55

下面的代码似乎完成了我所要求的操作:

代码语言:javascript
复制
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
票数 0
EN
查看全部 1 条回答
页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接:

https://stackoverflow.com/questions/60510856

复制
相关文章

相似问题

领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档