Python 多重处理函数，将 header 添加到列表中的 URL

Question

在 Python 中，我有一个函数，它接受 URL 列表(url_list)并添加 header 。该列表最多可以有 25,000 个 url，所以我想尝试使用多重处理。我尝试了以下代码，但我认为由于连接，它并没有真正进行多重处理。我怎样才能做到真正的多重处理？

def do_it(url_list, headers):
    for i in url_list:
        print "adding header to: \n" + i
        requests.post(i, headers=headers)
        print "done!"



  value = raw_input("Proceed? Enter [Y] for yes: ")
    if value == "Y":
        p = Process(target=do_it, args = (url_list, headers))
        p.start()
        p.join()
    else:
        print "Operation Failed!"

Answer 1

您绝对不想创建 25000 个子进程，每个进程处理 1 个 URL。但您也不希望 1 个进程处理 25000 个进程(这就是您构建的进程)。例如，您可能需要 8 个进程，每个进程处理 25000 的大约 1/8。

您可以通过创建一个充满 URL 的队列，创建 8 个进程来循环服务该队列，通过提取下一个 URL 并完成工作，然后加入所有 8 个进程来实现此目的。

但是您要做的是构建一个进程池。 multiprocessing 中已经内置了一个。除了已经构建和调试之外，它还具有您自己可能想不到的功能。它还可以以各种不同的方式传回结果。它可以让您一次将几个 URL 分批进行分块(如果您为每个进程提供完整列表的 1/8，则没有负载平衡；如果您一次为每个进程提供 1 个 URL，则您会在中间浪费时间) -可用于实际工作的流程通信)。等等。

所以，让我们使用它:

def do_it(url, headers):
    print "adding header to: \n" + i
    requests.post(i, headers=headers)
    print "done!"

pool = multiprocessing.Pool(max_workers=8)
results = pool.map(lambda url: do_it(url, headers), urls)
pool.join()

此代码唯一真正的问题是您正在等待它构建包含 25000 个结果的列表，而这些结果都是None。还有其他方法可以等待它而不构建返回结果，但实际上，此列表的成本不值得处理它的额外复杂性。

<小时/>

我使用lambda的原因是你需要一个只接受每个url的函数，并且你只有一个接受每个url<的函数 加上一个headers参数。您可以通过使用 lambda 或 def 定义新的包装器来创建该函数，或者通过调用为您完成此操作的高阶函数(例如 partial) >。这些基本上是等价的:

results = pool.map(lambda url: do_it(url, headers), urls)

def wrapper(url):
    return do_it(url, headers)
results = pool.map(wrapper, urls)

results = pool.map(partial(do_it, headers=headers), urls)

<小时/>

您可能还想考虑使用 Executor而不是一个普通的泳池。 Executor 返回更智能的结果对象，称为 Future，在许多情况下更容易处理。 (例如，您不必在 Pool 中的四种不同的 map 变体之间做出选择，您只需创建对简单的 submit 方法的理解即可，然后对生成的 Future 调用 as_completed 或 wait。)因为这直到 3.2 才添加到 Python，对于 2.x您必须安装向后移植库 futures使用它。无论如何，对于像这样的小案例，情况不会有太大不同:

with futures.ProcessPoolExecutor(max_workers=8) as executor:
    results = executor.map(lambda url: do_it(url, headers), urls)

...或:

with futures.ProcessPoolExecutor(max_workers=8) as executor:
    fs = [executor.submit(do_it, url, headers) for url in urls]
    futures.wait(fs)