在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) 个工作进程池。
• 有一个JoinableQueue，叫做url_queue。
• 每个工作人员从 中获取一个 url url_queue，找到新的 url 并将它们添加到 中url_queue。
• 仅在添加新项目后才会调用url_queue.task_done().
• 主进程调用url_queue.join(). 这会阻塞主进程，直到task_done为 url_queue.
• 由于工作进程的daemon属性设置为 True，因此它们也会在主进程结束时结束。

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