我是多处理的新手.我一直在阅读有关多处理模块的文档.我读过有关池,线程,队列等的内容,但我完全迷失了.
我想用多处理做的是,转换我简陋的http下载器,与多个工作者一起工作.我现在正在做的是,下载一个页面,解析到页面以获得有趣的链接.继续,直到下载所有有趣的链接.现在,我想用多处理来实现它.但我现在不知道如何组织这个工作流程.我有两个想法.首先,我想过有两个队列.需要下载的链接的一个队列,其他用于要解析的链接的队列.一名工作人员下载页面,并将其添加到队列中,该队列用于需要解析的项目.其他进程解析一个页面,并将它感兴趣的链接添加到另一个队列.我期望从这种方法出现的问题是：首先,为什么一次下载一页并一次解析页面.此外,一个进程如何知道在耗尽队列中的所有项目之后,有些项目要添加到队列中.
我想要使用的另一种方法是.有一个函数,可以用url作为参数调用.此函数下载文档并开始解析链接.每次遇到一个有趣的链接时,它会立即创建一个与自身运行相同功能的新线程.我对这种方法的问题是,我如何跟踪周围产生的所有进程,如何知道是否还有进程要运行.而且,我如何限制最大进程数.
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.
>每个工作人员从url_queue获取一个url,找到新的url并添加
他们到url_queue.
>只有在添加新项目之后才会调用url_queue.task_done().
>主进程调用url_queue.join().这阻止了主要
进程直到为每个任务调用task_done
url_queue.
>由于工作进程将守护程序属性设置为True,
当主要过程结束时,它们也会结束.
此示例中使用的所有组件也在Doug Hellman’s excellent Python Module of the Week tutorial on multiprocessing中进行了解释.