SimpleXMLRPCServer调用Celery任务

Question

我正在尝试用SimpleXMLRPCServer和Celery制作一个简单的RPC服务器。基本上，这个想法是远程客户端（client.py）可以通过xmlrpc.client向服务器（server.py）调用任务，其中包括注册为Celery任务（runnable.py）的函数

问题是，当RPC函数通过register_function注册时，我可以直接通过它的名称调用它，因此它将正确执行，但不使用Celery。我想实现的是通过client.py中的name.delay（）调用它，这是Celery执行它的方式，但不锁定服务器线程。因此，server.py应该像代理一样，允许多个客户端调用完整的函数集，如：

for task in flow:
    job = globals()[task]
    job.delay("some arg")
    while True:
        if job.ready():
            break

我曾尝试使用allow_dotted_names=True的register_instance，但我遇到了一个错误：

xmlrpc.client.Fault: <Fault 1: "<class 'TypeError'>:cannot marshal <class '_thread.RLock'> objects">

这让我想到了一个问题——是否有可能做这样的

简化代码：

服务器.py

# ...runnable.py import
# ...rpc init
def register_tasks():
    for task in get_all_tasks():
        setattr(self, task, globals()[task])
        self.server.register_function(getattr(self, task), task)

可运行.py

app = Celery("tasks", backend="amqp", broker="amqp://")
@app.task()
def say_hello():
    return "hello there"
@app.task()
def say_goodbye():
    return "bye, bye"
def get_all_tasks():
    tasks = app.tasks
    runnable = []
    for t in tasks:
        if t.startswith("modules.runnable"):
            runnable.append(t.split(".")[-1])
    return runnable

最后，client.py

s = xmlrpc.client.ServerProxy("http://127.0.0.1:8000")
print(s.say_hello())

Answer 1

我想出了一个主意，为Celery延迟函数创建一些额外的包装器。这些是以RPC客户端调用RPC.the_remote_task.delay（*args）的方式注册的。这会返回Celery作业ID，然后客户端通过RPC.ready（job_ID）询问作业是否准备好，并通过RPC.get（job_ID）获取结果。就目前而言，存在一个明显的安全漏洞，因为当你知道作业ID时，你可以获得结果，但仍然可以正常工作。

正在注册任务（server.py）

def register_tasks():
    for task in get_all_tasks():
        exec("""def """ + task + """_runtime_task_delay(*args):
return celery_wrapper(""" + task + """, "delay", *args)
setattr(self, task + "_delay", """ + task + """_runtime_task_delay)
            """)
        f_delay = task + "_delay"
        self.server.register_function(getattr(self, f_delay), task + ".delay")
    def job_ready(jid):
        return celery_wrapper(None, "ready", jid)
    def job_get(jid):
        return celery_wrapper(None, "get", jid)
    setattr(self, "ready", job_ready)
    setattr(self, "get", job_get)
    self.server.register_function(job_ready, "ready")
    self.server.register_function(job_get, "get")

包装器（server.py）

def celery_wrapper(task, method, *args):
    if method == "delay":
        job = task.delay(*args)
        job_id = job.id
        return job_id
    elif method == "ready":
        res = app.AsyncResult(args[0])
        return res.ready()
    elif method == "get":
        res = app.AsyncResult(args[0])
        return res.get()
    else:
        return "0"

和RPC调用（client.py）

jid = s.the_remote_task.delay("arg1", "arg2")
is_running = True
while is_running:
        is_running = not s.ready(jid)
        if not is_running:
                print(s.get(jid))
        time.sleep(.01)