python threading和multiprocessing模块基本用法实例分析

2023-08-18 16:20:06 82

本文实例讲述了pythonthreading和multiprocessing模块基本用法。分享给大家供大家参考，具体如下：

前言

这两天为了做一个小项目，研究了一下python的并发编程，所谓并发无非多线程和多进程，最初找到的是threading模块，因为印象中线程“轻量...”，“切换快...”，“可共享进程资源...”等等，但是没想到这里水很深，进而找到了更好的替代品multiprocessing模块。下面会讲一些使用中的经验。

后面出现的代码都在ubuntu10.04+python2.6.5的环境下测试通过。

一、使用threading模块创建线程

1、三种线程创建方式

（1）传入一个函数

这种方式是最基本的，即调用threading中的Thread类的构造函数，然后指定参数target=func，再使用返回的Thread的实例调用start()方法，即开始运行该线程，该线程将执行函数func，当然，如果func需要参数，可以在Thread的构造函数中传入参数args=(...)。示例代码如下：

#!/usr/bin/python
#-*-coding:utf-8-*-
importthreading
#用于线程执行的函数
defcounter(n):
cnt=0;
foriinxrange(n):
forjinxrange(i):
cnt+=j;
printcnt;
if__name__=='__main__':
#初始化一个线程对象，传入函数counter，及其参数1000
th=threading.Thread(target=counter,args=(1000,));
#启动线程
th.start();
#主线程阻塞等待子线程结束
th.join();

这段代码很直观，counter函数是一个很无聊的双重循环，需要注意的是th.join()这句，这句的意思是主线程将自我阻塞，然后等待th表示的线程执行完毕再结束，如果没有这句，运行代码会立即结束。join的意思比较晦涩，其实将这句理解成这样会好理解些“whileth.is_alive():time.sleep(1)”。虽然意思相同，但是后面将看到，使用join也有陷阱。

（2）传入一个可调用的对象

许多的python对象都是我们所说的可调用的，即是任何能通过函数操作符“（）”来调用的对象（见《python核心编程》第14章）。类的对象也是可以调用的，当被调用时会自动调用对象的内建方法__call__()，因此这种新建线程的方法就是给线程指定一个__call__方法被重载了的对象。示例代码如下：

#!/usr/bin/python
#-*-coding:utf-8-*-
importthreading
#可调用的类
classCallable(object):
def__init__(self,func,args):
self.func=func;
self.args=args;
def__call__(self):
apply(self.func,self.args);
#用于线程执行的函数
defcounter(n):
cnt=0;
foriinxrange(n):
forjinxrange(i):
cnt+=j;
printcnt;
if__name__=='__main__':
#初始化一个线程对象，传入可调用的Callable对象，并用函数counter及其参数1000初始化这个对象
th=threading.Thread(target=Callable(counter,(1000,)));
#启动线程
th.start();
#主线程阻塞等待子线程结束
th.join();

这个例子关键的一句是apply(self.func,self.args);这里使用初始化时传入的函数对象及其参数来进行一次调用。

（3）继承Thread类

这种方式通过继承Thread类，并重载其run方法，来实现自定义的线程行为，示例代码如下：

#!/usr/bin/python
#-*-coding:utf-8-*-
importthreading,time,random
defcounter():
cnt=0;
foriinxrange(10000):
forjinxrange(i):
cnt+=j;
classSubThread(threading.Thread):
def__init__(self,name):
threading.Thread.__init__(self,name=name);
defrun(self):
i=0;
whilei<4:
printself.name,'counting...\n';
counter();
printself.name,'finish\n';
i+=1;
if__name__=='__main__':
th=SubThread('thread-1');
th.start();
th.join();
print'alldone';

这个例子定义了一个SubThread类，它继承了Thread类，并重载了run方法，在方法中调用counter4次并打印一些信息，可以看到这种方式比较直观。在构造函数中要记得先调用父类的构造函数进行初始化。

2、python多线程的限制

python多线程有个讨厌的限制，全局解释器锁（globalinterpreterlock），这个锁的意思是任一时间只能有一个线程使用解释器，跟单cpu跑多个程序一个意思，大家都是轮着用的，这叫“并发”，不是“并行”。手册上的解释是为了保证对象模型的正确性！这个锁造成的困扰是如果有一个计算密集型的线程占着cpu，其他的线程都得等着....，试想你的多个线程中有这么一个线程，得多悲剧，多线程生生被搞成串行；当然这个模块也不是毫无用处，手册上又说了：当用于IO密集型任务时，IO期间线程会释放解释器，这样别的线程就有机会使用解释器了！所以是否使用这个模块需要考虑面对的任务类型。

二、使用multiprocessing创建进程

1、三种创建方式

进程的创建方式跟线程完全一致，只不过要将threading.Thread换成multiprocessing.Process。multiprocessing模块尽力保持了与threading模块在方法名上的一致性，示例代码可参考上面线程部分的。这里只给出第一种使用函数的方式：

#!/usr/bin/python
#-*-coding:utf-8-*-
importmultiprocessing,time
defrun():
i=0;
whilei<10000:
print'running';
time.sleep(2);
i+=1;
if__name__=='__main__':
p=multiprocessing.Process(target=run);
p.start();
#p.join();
printp.pid;
print'mastergone';

2、创建进程池

该模块还允许一次创建一组进程，然后再给他们分配任务。详细内容可参考手册，这部分研究不多，不敢乱写。

pool=multiprocessing.Pool(processes=4)
pool.apply_async(func,args...)

3、使用进程的好处

完全并行，无GIL的限制，可充分利用多cpu多核的环境；可以接受linux信号，后面将看到，这个功能非常好用。

三、实例研究

该实例假想的任务是：一个主进程会启动多个子进程分别处理不同的任务，各个子进程可能又有自己的线程用于不同的IO处理（前面说过，线程在IO方面还是不错的），要实现的功能是，对这些子进程发送信号，能被正确的处理，例如发生SIGTERM，子进程能通知其线程收工，然后“优雅”的退出。现在要解决的问题有：（1）在子类化的Process对象中如何捕捉信号；（2）如何“优雅的退出”。下面分别说明。

1、子类化Process并捕捉信号

如果是使用第一种进程创建方式（传入函数），那么捕捉信号很容易，假设给进程运行的函数叫func，代码示例如下：

#!/usr/bin/python
#-*-coding:utf-8-*-
importmultiprocessing,signal,time
defhandler(signum,frame):
print'signal',signum;
defrun():
signal.signal(signal.SIGTERM,handler);
signal.signal(signal.SIGINT,handler);
i=0;
whilei<10000:
print'running';
time.sleep(2);
i+=1;
if__name__=='__main__':
p=multiprocessing.Process(target=run);
p.start();
#p.join();
printp.pid;
print'mastergone';

这段代码是在第一种创建方式的基础上修改而来的，增加了两行signal.signal(...)调用，这是说这个函数要捕捉SIGTERM和SIGINT两个信号，另外增加了一个handler函数，该函数用于捕捉到信号时进行相应的处理，我们这里只是简单的打印出信号值。

注意p.join()被注释掉了，这里跟线程的情况有点区别，新的进程启动后就开始运行了，主进程也不用等待它运行完，可以该干嘛干嘛去。这段代码运行后会打印出子进程的进程id，根据这个id，在另一个终端输入kill-TERMid，会发现刚才的终端打印出了"signal15"。

但是使用传入函数的方式有一点不好的是封装性太差，如果功能稍微复杂点，将会有很多的全局变量暴露在外，最好还是将功能封装成类，那么使用类又怎么注册信号相应函数呢？上面的例子貌似只能使用一个全局的函数，手册也没有给出在类中处理信号的例子，其实解决方法大同小异，也很容易，这个帖子http://stackoverflow.com/questions/6204443/python-signal-reading-return-from-signal-handler-function给了我灵感：

classMaster(multiprocessing.Process):
def__init__(self):
super(Master,self).__init__();
signal.signal(signal.SIGTERM,self.handler);#注册信号处理函数
self.live=1;
#信号处理函数
defhandler(self,signum,frame):
print'signal:',signum;
self.live=0;
defrun(self):
print'PID:',self.pid;
whileself.live:
print'living...'
time.sleep(2);

方法很直观，首先在构造函数中注册信号处理函数，然后定义了一个方法handler作为处理函数。这个进程类会每隔2秒打印一个“living...”，当接收到SIGTERM后，改变self.live的值，run方法的循环检测到这个值为0后就结束了，进程也结束了。

2、让进程优雅的退出

下面放出这次的假想任务的全部代码，我在主进程中启动了一个子进程（通过子类化Process类），然后子进程启动后又产生两个子线程，用来模拟“生产者-消费者”模型，两个线程通过一个队列进行交流，为了互斥访问这个队列，自然要加一把锁（condition对象跟Lock对象差不多，不过多了等待和通知的功能）；生产者每次产生一个随机数并扔进队列，然后休息一个随机时间，消费者每次从队列取一个数；而子进程中的主线程要负责接收信号，以便让整个过程优雅的结束。代码如下：

#!/usr/bin/python
#-*-coding:utf-8-*-
importtime,multiprocessing,signal,threading,random,time,Queue
classMaster(multiprocessing.Process):
def__init__(self):
super(Master,self).__init__();
signal.signal(signal.SIGTERM,self.handler);
#这个变量要传入线程用于控制线程运行，为什么用dict？充分利用线程间共享资源的特点
#因为可变对象按引用传递，标量是传值的，不信写成self.live=true试试
self.live={'stat':True};
defhandler(self,signum,frame):
print'signal:',signum;
self.live['stat']=0;#置这个变量为0，通知子线程可以“收工”了
defrun(self):
print'PID:',self.pid;
cond=threading.Condition(threading.Lock());#创建一个condition对象，用于子线程交互
q=Queue.Queue();#一个队列
sender=Sender(cond,self.live,q);#传入共享资源
geter=Geter(cond,self.live,q);
sender.start();#启动线程
geter.start();
signal.pause();#主线程睡眠并等待信号
whilethreading.activeCount()-1:#主线程收到信号并被唤醒后，检查还有多少线程活着（除掉自己）
time.sleep(2);#再睡眠等待，确保子线程都安全的结束
print'checkinglive',threading.activeCount();
print'matergone';
classSender(threading.Thread):
def__init__(self,cond,live,queue):
super(Sender,self).__init__(name='sender');
self.cond=cond;
self.queue=queue;
self.live=live
defrun(self):
cond=self.cond;
whileself.live['stat']:#检查这个进程内的“全局”变量，为真就继续运行
cond.acquire();#获得锁，以便控制队列
i=random.randint(0,100);
self.queue.put(i,False);
ifnotself.queue.full():
print'senderadd:',i;
cond.notify();#唤醒等待锁的其他线程
cond.release();#释放锁
time.sleep(random.randint(1,3));
print'senderdone'
classGeter(threading.Thread):
def__init__(self,cond,live,queue):
super(Geter,self).__init__(name='geter');
self.cond=cond;
self.queue=queue;
self.live=live
defrun(self):
cond=self.cond;
whileself.live['stat']:
cond.acquire();
ifnotself.queue.empty():
i=self.queue.get();
print'geterget:',i;
cond.wait(3);
cond.release();
time.sleep(random.randint(1,3));
print'geterdone'
if__name__=='__main__':
master=Master();
master.start();#启动子进程

需要注意的地方是，在Master的run方法中sender.start()和geter.start()之后，按常理应该接着调用sender.join()和geter.join()，让主线程等待子线程结束，前面说的join的陷阱就在这里，join将主线程阻塞（blocking）住了，主线程无法再捕捉信号，刚开始研究这块时还以为信号处理函数写错了。网上讨论比较少，这里说的比较清楚http://stackoverflow.com/questions/631441/interruptible-thread-join-in-python，http://www.gossamer-threads.com/lists/python/python/541403

参考：

《python核心编程》
《pythonmanual》

更多关于Python相关内容感兴趣的读者可查看本站专题：《Python进程与线程操作技巧总结》、《Python数据结构与算法教程》、《Python函数使用技巧总结》、《Python字符串操作技巧汇总》、《Python入门与进阶经典教程》、《Python+MySQL数据库程序设计入门教程》及《Python常见数据库操作技巧汇总》

希望本文所述对大家Python程序设计有所帮助。