Python中编写并发程序

GIL

在Python中,因为历史缘由(GIL),使得Python中多线程的效果很是不理想.GIL使得任什么时候刻Python只能利用一个CPU核,而且它的调度算法简单粗暴:多线程中,让每一个线程运行一段时间t,而后强行挂起该线程,继而去运行其余线程,如此周而复始,直到全部线程结束.

这使得没法有效利用计算机系统中的"局部性",频繁的线程切换也对缓存不是很友好,形成资源的浪费.

听说Python官方曾经实现了一个去除GIL的Python解释器,可是其效果还不若有GIL的解释器,遂放弃.后来Python官方推出了"利用多进程替代多线程"的方案,在Python3中也有concurrent.futures这样的包,让咱们的程序编写能够作到"简单和性能兼得".

多进程/多线程+Queue

通常来讲,在Python中编写并发程序的经验是:计算密集型任务使用多进程,IO密集型任务使用多进程或者多线程.另外,由于涉及到资源共享,因此须要同步锁等一系列麻烦的步骤,代码编写不直观.另一种好的思路是利用多进程/多线程+Queue的方法,能够避免加锁这样麻烦低效的方式.

如今在Python2中利用Queue+多进程的方法来处理一个IO密集型任务.
假设如今须要下载多个网页内容并进行解析,单进程的方式效率很低,因此使用多进程/多线程势在必行.
咱们能够先初始化一个tasks队列,里面将要存储的是一系列dest_url,同时开启4个进程向tasks中取任务而后执行,处理结果存储在一个results队列中,最后对results中的结果进行解析.最后关闭两个队列.

下面是一些主要的逻辑代码.

python# -*- coding:utf-8 -*-

#IO密集型任务
#多个进程同时下载多个网页
#利用Queue+多进程
#因为是IO密集型,因此一样能够利用threading模块

import multiprocessing

def 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()

利用Python3中的concurrent.futures包

在Python3中能够利用concurrent.futures包,编写更加简单易用的多线程/多进程代码.其使用感受和Java的concurrent框架很类似(借鉴?)
好比下面的简单代码示例

pythondef 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

总结

要是一些大型Python项目也这般编写,那么效率也过低了.在Python中有许多已有的框架使用,使用它们起来更加高效.
可是本身的一些"小打小闹"的程序这样来编写仍是不错的.:)