Python中编写并发程序

GIL

在Python中,由于历史原因(GIL),使得Python中多线程的效果非常不理想.GIL使得任何时刻Python只能利用一个CPU 核,并且它的调度算法简单粗暴:多线程中,让每个线程运行一段时间t,然后强行挂起该线程,继而去运行其他线程,如此周而复始,直到所有线程结束.

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

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

多进程/多线程+Queue

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

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

# -*- 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框架很相似(借鉴?)比如下面的简单代码示例

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

总结

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