python 中的多处理使用哪种策略

Question

我对多处理完全陌生。我一直在阅读有关多处理模块的文档。我读过有关池、线程、队列等的内容，但我完全迷失了。

我想要对多处理做的是，将我简陋的 http 下载器转换为与多个工作人员一起工作。我现在正在做的是，下载一个页面，解析页面以获取有趣的链接。继续，直到下载所有有趣的链接。现在，我想通过多处理来实现这一点。但我目前不知道如何组织这个工作流程。对此我有两个想法。首先，我考虑有两个队列。一个队列用于需要下载的链接，另一个用于要解析的链接。一名工作人员下载页面，并将它们添加到队列中，该队列用于需要解析的项目。其他进程解析页面，并将其发现有趣的链接添加到其他队列。我预计这种方法会出现以下问题：首先，为什么要一次下载一个页面并一次解析一个页面。此外，在耗尽队列中的所有项目后，一个进程如何知道稍后有项目要添加到队列中。

我考虑使用的另一种方法是这样的。有一个函数，可以使用 url 作为参数来调用。此函数下载文档并开始解析其中的链接。每次遇到有趣的链接时，它都会立即创建一个新线程，运行与自身相同的功能。这种方法的问题是，如何跟踪周围产生的所有进程，如何知道是否仍有进程正在运行。另外，如何限制最大进程数。

所以我完全迷失了。任何人都可以提出一个好的策略，也许还可以展示一些关于如何实现这个想法的示例代码。

Answer 1

这是一种使用多处理的方法。 （非常感谢@Voo，对代码提出了许多改进建议）。

import multiprocessing as mp
import logging
import Queue
import time

logger=mp.log_to_stderr(logging.DEBUG)  # or, 
# logger=mp.log_to_stderr(logging.WARN) # uncomment this to silence debug and info messages

def worker(url_queue,seen):
    while True:
        url=url_queue.get()
        if url not in seen:
            logger.info('downloading {u}'.format(u=url))
            seen[url]=True
            # Replace this with code to dowload url
            # urllib2.open(...)
            time.sleep(0.5)
            content=url
            logger.debug('parsing {c}'.format(c=content))
            # replace this with code that finds interesting links and
            # puts them in url_queue
            for i in range(3):
                if content<5:
                    u=2*content+i-1
                    logger.debug('adding {u} to url_queue'.format(u=u))
                    time.sleep(0.5)
                    url_queue.put(u)
        else:
            logger.debug('skipping {u}; seen before'.format(u=url))
        url_queue.task_done()

if __name__=='__main__':
    num_workers=4
    url_queue=mp.JoinableQueue()
    manager=mp.Manager()
    seen=manager.dict()

    # prime the url queue with at least one url
    url_queue.put(1)
    downloaders=[mp.Process(target=worker,args=(url_queue,seen))
                 for i in range(num_workers)]
    for p in downloaders:
        p.daemon=True
        p.start()
    url_queue.join()

• 创建 (4) 个工作进程池。
• 有一个名为url_queue的JoinableQueue。
• 每个worker从url_queue获取一个url，找到新的url并添加
  将它们添加到 url_queue 中。
• 仅在添加新项目后，它才会调用 url_queue.task_done()。
• 主进程调用url_queue.join()。这会阻止主要
  处理直到为每个任务调用 task_done
url_queue。
• 由于工作进程的 daemon 属性设置为 True，
  当主进程结束时它们也结束。

此示例中使用的所有组件也在 Doug Hellman 的本周优秀 Python 模块教程中进行了解释多处理。

Answer 2

您所描述的本质上是图遍历；大多数图遍历算法（比深度优先更复杂）都会跟踪两组节点，在您的情况下，节点是 url 的。

第一个集合称为“闭集”，表示所有已被访问和处理的节点。如果在处理页面时发现某个链接恰好位于封闭集中，则可以忽略它，因为它已经被处理了。

第二组毫不奇怪地被称为“开集”，并且包括已找到但尚未处理的所有边。

基本机制是首先将根节点放入开集（闭集最初为空，尚未处理任何节点），然后开始工作。每个工作进程从开放集中获取一个节点，将其复制到封闭集中，处理该节点，并将其发现的任何节点添加回开放集中（只要它们不在开放集中或封闭集中） 。一旦开集为空（并且没有工作人员仍在处理节点），图就已被完全遍历。

实际上在多处理中实现这一点可能意味着您将有一个主任务来跟踪开集和闭集；如果工作池中的工作线程表明它已准备好工作，则主工作线程负责将节点从开放集移动到封闭集并启动该工作线程。然后，工作人员可以将他们找到的所有节点传递回主节点，而不必担心它们是否已经关闭；并且 master 将忽略已经关闭的节点。