简单实现并发：python concurrent模块

trzxycx

转载：http://www.bubuko.com/infodetail-167068.html


可以使用python 3中的concurrent模块， 如果python环境是2.7的话，需要下载https://pypi.python.org/packages/source/f/futures /futures-2.1.6.tar.gz#md5=cfab9ac3cd55d6c7ddd0546a9f22f453

此futures包即可食用concurrent模块。　　官方文档：http://pythonhosted.org//futures/
　　


对于python来说，作为解释型语言，Python的解释器必须做到既安全又高效。我们都知道多线程编程会遇到的问题，解释器要留意的是避免在不同的线程操作内部共享的数据，同时它还要保证在管理用户线程时保证总是有最大化的计算资源。而python是通过使用全局解释器锁来保护数据的安全性：

python 代码的执行由python虚拟机来控制，即Python先把代码（.py文件）编译成字节码（字节码在Python虚拟机程序里对应的是 PyCodeObject对象，.pyc文件是字节码在磁盘上的表现形式），交给字节码虚拟机，然后虚拟机一条一条执行字节码指令，从而完成程序的执行。 python在设计的时候在虚拟机中，同时只能有一个线程执行。同样地，虽然python解释器中可以运行多个线程，但在任意时刻，只有一个线程在解释器 中运行。而对python虚拟机的访问由全局解释器锁来控制，正是这个锁能保证同一时刻只有一个线程在运行。在多线程的环境中，python虚拟机按一下 方式执行：

1，设置GIL(global interpreter lock).

2，切换到一个线程执行。

3，运行：

    a，指定数量的字节码指令。

    b，线程主动让出控制（可以调用time.sleep(0)）。

4，把线程设置为睡眠状态。

5，解锁GIL.

6，再次重复以上步骤。

GIL的特性，也就导致了python不能充分利用多核cpu。而 对面向I/O的（会调用内建操作系统C代码的）程序来说，GIL会在这个I/O调用之前被释放，以允许其他线程在这个线程等待I/O的时候运行。如果线程 并为使用很多I/O操作，它会在自己的时间片一直占用处理器和GIL。这也就是所说的：I/O密集型python程序比计算密集型的程序更能充分利用多线 程的好处。

总之，不要使用python多线程，使用python多进程进行并发编程，就不会有GIL这种问题存在，并且也能充分利用多核cpu。



一，提供的功能

提供了多线程和多进程的并发功能

二，基本方法

class   concurrent.futures.Executor (注：Executor为ThreadPoolExecutor或者ProcessPoolExecutor)

提供的方法如下：

    submit(fn, *args, **kwargs)

    fn：为需要异步执行的函数

    args，kwargs：为给函数传递的参数

    例：

#!/bin/env python
#coding:utf-8
import time,re
import os,datetime
from concurrent import futures
def wait_on_b():
    print 5
    time.sleep(2)
def wait_on_a():
    print 6
    time.sleep(2)

ex = futures.ThreadPoolExecutor(max_workers=2)
ex.submit(wait_on_b)
ex.submit(wait_on_a)

wait_on_a和wait_on_b函数会同时执行，因为使用了2个worker

#####################################

     map(func, *iterables, timeout=None)


此map函数和python自带的map函数功能类似，只不过concurrent模块的map函数从迭代器获得参数后异步执行。并且，每一个异步操作，
能用timeout参数来设置超时时间，timeout的值可以是int或float型，如果操作timeout的话，会raisesTimeoutError。如果timeout参数不指定的话，则不设置超时间。

     func：为需要异步执行的函数

     iterables：可以是一个能迭代的对象，例如列表等。每一次func执行，会从iterables中取参数。

     timeout：设置每次异步操作的超时时间

     例：

#!/bin/env python
#coding:utf-8
import time,re
import os,datetime
from concurrent import futures
data = [‘1‘,‘2‘]
def wait_on(argument):
    print argument
    time.sleep(2)
    return ‘ok‘
ex = futures.ThreadPoolExecutor(max_workers=2)
for i in ex.map(wait_on,data):
    print i

map函数异步执行完成之后，结果也是list，数据需要从list中取出

######################################

submit函数和map函数，根据需要，选一个使用即可。

     shutdown(wait=True)

     此函数用于释放异步执行操作后的系统资源。


    If wait is True then this method will not return until all the
pending futures are done executing and the resources associated with the
executor have been freed. If wait is False then this method will return
immediately and the resources associated with the executor will be
freed when all pending futures are done executing. Regardless of the
value of wait, the entire Python program will not exit until all pending
futures are done executing.

You
can avoid having to call this method explicitly if you use the with
statement, which will shutdown the Executor (waiting as if
Executor.shutdown() were called with wait set to True):


with ThreadPoolExecutor(max_workers=4) as e:
e.submit(shutil.copy, ‘src1.txt‘, ‘dest1.txt‘)


三，完整的concurrent例子：

#!/bin/env python
#coding:utf-8
import time,re,fcntl
import os,datetime
from concurrent import futures
count_list = list()
MinuteNum = 1
StartTime = datetime.datetime(2014, 4, 16, 19, 31, 0, 484870)
NowTime = datetime.datetime.now()
os.system(‘:>new.txt‘)
f_new = open(‘new.txt‘,‘a‘)
def test(CountTimeFormat):
    f = open(‘push_slave.stdout‘,‘r‘)
    for line in f.readlines():
        if re.search(CountTimeFormat,line):

            #获得文件专用锁
            fcntl.flock(f_new, fcntl.LOCK_EX)
            f_new.writelines(line)
            f_new.flush()

            #释放文件锁
            fcntl.flock(f_new, fcntl.LOCK_UN)
            break
while 1:
    AfterOneMinute = datetime.timedelta(minutes=MinuteNum)
    CountTime = AfterOneMinute+StartTime
    CountTimeFormat = CountTime.strftime(‘%Y-%m-%d %H:%M‘)
    MinuteNum = MinuteNum+1
    count_list.append(CountTimeFormat)
    if CountTimeFormat == "2014-04-23 16:00":
        break
def exec_cmd():
    with futures.ProcessPoolExecutor(max_workers=24) as executor:
        dict(( executor.submit(test, times), times) for times in count_list)
if __name__ == ‘__main__‘:
    exec_cmd()
    f_new.close()python concurrent模块,布布扣,bubuko.com　　简单实现并发：python concu