如何重复请求任务答案

【问题标题】：How to make repeated requests for tasks如何重复请求任务
【发布时间】：2019-11-09 17:20:56
【问题描述】：

我想向服务器发出重复请求，该服务器将返回一些任务。服务器的响应将是一个字典，其中包含需要调用的函数列表。例如：

{ 
   tasks: [
      {
         function: "HelloWorld",
         id: 1212
      },
      {
         function: "GoodbyeWorld"
         id: 1222
      }
   ]
}

注意：我正在模拟它。

对于这些任务中的每一个，我将使用multiprocessing 运行指定的函数。这是我的代码示例：

r = requests.get('https://localhost:5000', auth=('user', 'pass'))
data = r.json()

if len(data["tasks"]) > 0:
  manager = multiprocessing.Manager()
  for task in data["tasks"]:
    if task["function"] == "HelloWorld":
      helloObj = HelloWorldClass()
      hello = multiprocessing.Process(target=helloObj.helloWorld)
      hello.start()
      hello.join()
    elif task["function"] == "GoodbyeWorld":
      byeObj = GoodbyeWorldClass()
      bye = multiprocessing.Process(target=byeObj.byeWorld)
      bye.start()
      bye.join()

问题是，当其他进程正在运行时，我想重复请求并填充data["tasks"] 数组。如果我将所有内容都放入某个 while 循环中，它只会在初始响应中的所有进程完成后发出请求（当所有进程都达到 join() 时）。

谁能帮我提出重复请求并不断填充数组？如果我需要做任何澄清，请告诉我。

【问题讨论】：

看起来你需要的是一个队列。首先，您将所有响应结果放到其中一个docs.python.org/3/library/queue.html 中，然后一个一个地弹出队列中的项目并开始处理直到完成。 HTTP 响应不会像处理数组那样干扰队列，而且代码会更容易推理。
@jim 你考虑过使用 asyncio 还是 Celery 吗？或者这不是你的选择？
所以基本上你打电话给start()，然后在你等待这个线程以join()结束之后。您应该列出一些threads=[]，然后在创建threads += [hello] 时添加它，再见。最后加入for th in threads: th.join()

标签： python python-3.x asynchronous python-multiprocessing

【解决方案1】：

如果我理解正确，你需要这样的东西：

import time
from multiprocessing import Process

import requests

from task import FunctionFactory


def get_tasks():
    resp = requests.get('https://localhost:5000', auth=('user', 'pass'))
    data = resp.json()

    return data['tasks']


if __name__ == '__main__':
    procs = {}

    for _ in range(10):
        tasks = get_tasks()

        if not tasks:
            time.sleep(5)
            continue

        for task in tasks:

            if task['id'] in procs:
                # This task has been already submitted for execution.
                continue

            func = FunctionFactory.build(task['function'])

            proc = Process(target=func)
            proc.start()

            procs[task['id']] = proc

    # Waiting for all the submitted tasks to finish.
    for proc in procs.values():
        proc.join()

这里，函数get_tasks 用于从服务器请求具有id 和function 键的字典列表。在主要部分中，有一个 procs 字典将 id 映射到正在运行的流程实例，这些流程实例使用接收到的任务的 function 名称执行由 FunctionFactory 构建的函数。如果已经有一个正在运行的具有相同 id 的任务，它将被忽略。

使用这种方法，您可以根据需要经常请求任务（这里，10 请求用于for 循环）并启动进程以并行执行它们。最后，您只需等待所有提交的任务完成即可。

【讨论】：

【解决方案2】：

您的程序有错误，您应该在创建所有任务之后调用joins。加入块直到过程完成 - 在你开始下一个之前。这实际上使您整个程序按顺序运行。

【讨论】：