# -*- coding:utf-8 -*-#IO密集型任務#多個進程同時下載多個網頁#利用Queue+多進程#由于是IO密集型,所以同樣可以利用threading模塊import multiprocessingdef main():  tasks = multiprocessing.JoinableQueue()  results = multiprocessing.Queue()  cpu_count = multiprocessing.cpu_count() #進程數目==CPU核數目  create_process(tasks, results, cpu_count)  #主進程馬上創建一系列進程,但是由于阻塞隊列tasks開始為空,副進程全部被阻塞  add_tasks(tasks) #開始往tasks中添加任務  parse(tasks, results) #最后主進程等待其他線程處理完成結果def create_process(tasks, results, cpu_count):  for _ in range(cpu_count):    p = multiprocessing.Process(target=_worker, args=(tasks, results)) #根據_worker創建對應的進程    p.daemon = True #讓所有進程可以隨主進程結束而結束    p.start() #啟動def _worker(tasks, results):  while True:  #因為前面所有線程都設置了daemon=True,故不會無限循環    try:      task = tasks.get()  #如果tasks中沒有任務,則阻塞      result = _download(task)      results.put(result)  #some exceptions do not handled    finally:      tasks.task_done()def add_tasks(tasks):  for url in get_urls(): #get_urls() return a urls_list    tasks.put(url)def parse(tasks, results):  try:     tasks.join()  except KeyboardInterrupt as err:    print "Tasks has been stopped!"    print err  while not results.empty():    _parse(results)if __name__ == '__main__':  main()

def handler():  futures = set()  with concurrent.futures.ProcessPoolExecutor(max_workers=cpu_count) as executor:    for task in get_task(tasks):      future = executor.submit(task)      futures.add(future)def wait_for(futures):  try:    for future in concurrent.futures.as_completed(futures):      err = futures.exception()      if not err:        result = future.result()      else:        raise err  except KeyboardInterrupt as e:    for future in futures:      future.cancel()    print "Task has been canceled!"    print e  return result