Python核心编程：第18章 多线程编程

18.1 引言/动机

1.多线程编程对于某些任务是最理想的。这些任务具有以下特点：它们本质就是异步的，需要有多个并发事务，各个事务的运行顺序可以是不确定的、随机的、

不可预测的。这样的编程任务可以被分成多个执行流，每个流都有一个要完成的目标。根据应用的不同，这些子任务可能都要计算出一个中间结果，用于合并得到最后

的结果。

一个顺序执行的程序要从每个I/O（输入/输出）终端信道检查用户的输入时，程序无论如何也不在I/O终端通道的时候阻塞。顺序执行的程序必须使用非阻塞I/O，或是带有计时器

的阻塞I/O

使用多线程和一个共享的数据结构如Queue，这种程序任务可以用功能单一的线程来组织。

UserRequestThread:负责读取客户的输入，可能是一个I/O通道。程序可能创建多个线程，每个客户一个，请求将被放入队列

RequestProcessor:一个负责从队列中获取并处理请求的线程，它为下面那种线程提供输出

ReplyThread:负责把给用户的输出取出来，如果是网络应用程序就把结果发送出去，否则就保存到本地文件系统或数据中。

18.2 线程和进程

18.2.1 什么是进程

进程：进程是程序的一次执行。每个进程都有自己的地址空间、内存、数据栈及其他记录其运行轨迹的辅助数据。

各个进程间使用进程间通讯（interprocess communication， IPC），而不能直接共享信息。

18.2.2 什么是线程

所有的线程运行在同一个进程中，共享相同的运行环境。

线程有三部分：开始，顺序执行和结束部分。线程的运行可能被抢占（中断），或暂时的被挂起（也叫睡眠），让其他的线程运行，这叫做让步。

一个进程中的各个线程之间共享的同一步数据空间，所以线程之间可以比进程之间更方便地共享数据以及相互通讯。

多个线程共同访问同一片数据，由于数据访问的顺序不一样，有可能导致数据结果的不一致问题。这叫做竞态条件。

有的函数会在完成之前阻塞住，在没有特别为多线程修改的清空下，这种贪婪的函数会让CPU的时间分配有所倾斜。导致各个线程分配到的运行时间尽可能不尽相同，不尽公平。

18.3 Python、线程和全局解释器锁

18.3.1 全局解释器锁（GIL)

Python解释器中可以”运行“多个线程，但在任意时刻，只有一个进程在解释器中运行。

对Python虚拟机的访问是由全局解释器锁（global interpreter lock ,GIL)来控制，正是这个锁能保证同一时刻只有一个进程在运行。

1.设置GIL

2.切换到一个线程去运行

3.运行：

a.指定数量的字节码的指令，或者

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

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

5.解锁GIL

6.重复

18.3.2 退出线程

当一个线程结束计算，它就退出了。线程可以调用thread.exit()之类的退出函数，也可以使用Python退出进程的标准方法，如sys.exit()或抛出一个SystemExit异常等。

18.3.4 在Python中使用进程

#-*-coding:utf-8-*-
from time import sleep,ctime

def loop0():
print 'start loop 0 at :',ctime()
sleep(4)
print'loop o done at:', ctime()

def loop1():
print 'start loop 1 at:',ctime()
sleep(2)
print 'loop 1 done at:',ctime()

def main():
print 'starting at:',ctime()
loop0()
loop1()
print 'all DONE at:',ctime()

if __name__ =='__main__':
main()

18.3.5 Python中的threading模块

thread模块提供了基本的线程和锁的支持，而threading提供了更高级别，功能更强的线程管理功能。

避免使用thread模块

18.4 thread 模块

start_new_thread 产生一个新的进程，在新进程中用指定的参数和可选的kwargs来调用这个参数

allocate_lock() 分配一个LockType 类型的锁对象

acquire() 尝试获取锁对象

locked() 如果获取了锁对象返回True，否则返回False

relase() 释放锁

#-*-coding:utf-8-*-
import thread
from time import sleep,ctime

def loop0():
print 'start loop 0 at :',ctime()
sleep(4)
print'loop o done at:', ctime()

def loop1():
print 'start loop 1 at:',ctime()
sleep(2)
print 'loop 1 done at:',ctime()

def main():
print 'starting at:',ctime()
thread.start_new_thread(loop0,())
thread.start_new_thread(loop1,())
sleep(6)
print 'all DONE at:',ctime()

if __name__ =='__main__':
main()

为了防止主线程关闭loop(0)和loop(1),加了一个sleep(6)

使用线程和锁

#-*-coding:utf-8-*-
import thread
from time import sleep,ctime

loops = [4,2]
def loop(nloop,nsec,lock):
print 'start loop', nloop, 'at:',ctime()
sleep(nsec)
print 'loop',nloop,'done at:',ctime()
lock.release()

def main():
print'starting at:',ctime()
locks = []
nloops = range(len(loops))
for i in nloops:
lock = thread.allocate_lock()
lock.acquire()
locks.append(lock)

for i in nloops:
thread.start_new_thread(loop,(i,loops[i],locks[i]))

for i in nloops:
while locks[i].locked():pass

print 'all done at:', ctime()

if __name__ == '__main__':
main()

18.5 threading模块

threading模块对象 描述

Thread 表示一个线程的执行对象

Lock 锁原语对象

Rlock 可重锁对象

Codndition 条件变量对象能让一个线程停下来，等待其他线程满足了某个‘条件’。

另一个避免使用thread模块的原因是，它不支持守护线程

18.5.1 Thread类

threading的Thread类是你主要的运行对象。

用Thread类，可以使用多种方法来创建线程

1.创建一个Thread的实例，传给它一个函数

2.创建一个Thread的实例，传给它一个可调用的类的对象

3.从Thread派生出一个子类，创建一个这个子类的实例

start() 开始线程的执行

run() 定义线程的功能的函数

join(time=None) 程序挂起，直到线程结束;如果给了timeout,则最多阻塞timeout秒

getName() 返回线程的名字

setName 设置线程的名字

isAlive() 布尔标志，表示这个线程是否还在运行中

isDaemon 返回线程的Daemon标志

setDaemon 把线程的Daemon标志改为daemonic

创建一个Thread的实例，传给它一个函数：

实例化一个Thread(调用Thread())与调用thread.start_new_thread()之间最大的区别就是，新的线程不会立即开始。在创建线程对象，但不想马上开始运行线程的时候，这是一个很有用的同步特性

#-*-coding:utf-8-*-
import threading
from time import sleep,ctime

loops = [4,2]
def loop(nloop,nsec,):
print 'start loop', nloop, 'at:',ctime()
sleep(nsec)
print 'loop',nloop,'done at:',ctime()

def main():
print'starting at:',ctime()
threads=[]
nloops = range(len(loops))
for i in nloops:
t=threading.Thread(target=loop,args=(i,loops[i]))
threads.append(t)

for i in nloops:
threads[i].start()
for i in nloops:
threads[i].join()

print 'all done at:', ctime()

if __name__ == '__main__':
main()

join()会等到线程结束，或者在给了timeout参数的时候，等到超时为止。

join()的另一个比较重要的方面是它可以完全不用调用。一旦线程启动后，就会一直运行，直到线程的函数结束，退出为止。如果主线程除了等待线程结束外，还有其他的事情要做，那就不用调用join(),只有在你要等待线程结束的时候才调用join()

创建一个Thread()的实例，传给它一个可调用的类的对象。

优点：这是多线程编程的一个更为面向对象的方法。相对于一个或几个函数来说，由于类对象里可以使用类的强大的功能，可以保存更多的信息，这种方法更为灵活。

#-*-coding:utf-8-*-
import threading
from time import sleep,ctime

loops = [4,2]

class ThreadFunc(object):
def __init__(self,func,args,name=''):
self.name=name
self.func=func
self.args=args
def __call__(self):
apply(self.func,self.args)
def loop(nloop,nsec,):
print 'start loop', nloop, 'at:',ctime()
sleep(nsec)
print 'loop',nloop,'done at:',ctime()

def main():
print'starting at:',ctime()
threads=[]
nloops = range(len(loops))
for i in nloops:
t=threading.Thread(target=ThreadFunc(loop,(i,loops[i]),loop.__name__))

threads.append(t)

for i in nloops:
threads[i].start()
for i in nloops:
threads[i].join()

print 'all done at:', ctime()

if __name__ == '__main__':
main()

从Thread派生出一个子类，创建一个这个子类的实例。可以更灵活地定制我们的线程对象，而且在创建线程的时候也更简单。

#-*-coding:utf-8-*-
import threading
from time import sleep,ctime

loops = [4,2]

class MyThread(threading.Thread):
def __init__(self,func,args,name=''):
threading.Thread.__init__(self)
self.name=name
self.func=func
self.args=args
def run(self):
apply(self.func,self.args)

def loop(nloop,nsec,):
print 'start loop', nloop, 'at:',ctime()
sleep(nsec)
print 'loop',nloop,'done at:',ctime()

def main():
print'starting at:',ctime()
threads=[]
nloops = range(len(loops))
for i in nloops:
t=MyThread(loop,(i,loops[i]),loop.__name__)
threads.append(t)

for i in nloops:
threads[i].start()
for i in nloops:
threads[i].join()

print 'all done at:', ctime()

if __name__ == '__main__':
main()

斐波那契、阶乘和累加和

#-*-coding:utf-8-*-
import threading
from time import sleep,ctime

loops = [4,2]

class MyThread(threading.Thread):
def __init__(self,func,args,name=''):
threading.Thread.__init__(self)
self.name=name
self.func=func
self.args=args
def getResult(self):
return self.res
def run(self):
print'starting', self.name,'at:',\
ctime()
self.res = apply(self.func,self.args)
print self.name, 'finished at:',\
ctime()
def fib(x):
sleep(0.005)
if x<2:
return 1
else:
return (fib(x-2)+fib(x-1))

def fac(x):
sleep(0.1)
if x<2:
return 1
else:
return (x*sum(x-1))
def sum(x):
sleep(0.1)
if x<2:
return 1
else:
return  (x+sum(x-1))

funcs = [fib,fac,sum]
n = 12
def main():
nfuncs = range(len(funcs))
print '*** SINGLE THREAD'
for i in nfuncs:
print 'starting', funcs[i].__name__,'at:',\
ctime()
print funcs[i](n)
print funcs[i].__name__,'finished at:',\
ctime()
print '\n*** MUlTIPLE THREADS'
threads = []
for i in nfuncs:
t = MyThread(funcs[i],(n,),funcs[i].__name__)
threads.append(t)

for i in nfuncs:
threads[i].start()

for i in nfuncs:
threads[i].join()
print threads[i].getResult()

print 'all DONE'

if __name__ =='__main__':
main()

18.5.3 threading模块中的其他函数

18.5.4 生产者-消费者问题和Queue模块

生产者生产货物，然后把货物放到一个队列之类的数据结构中，生产货物所需要的时间无法预先确定。消费者消耗生产者生产货物的时间也是不确定的

Queue模块可以用来进行线程间通讯，让各个线程之间共享数据。

Queue模块函数

queue(sizez) 创建一个大小为size的Queue对象

Queue对象函数

qsize() 返回队列的大小（由于在返回的时候，队列可能会其他线程修改，所以这个值是近似值）

empty() 如果队列为空返回True,否则返回False

full() 如果队列已满返回True，否则返回False

put(item,block=0) 把item放到队列中，如果给了block(不为0)，函数会一直阻塞到队列中有空间为止

get(block=0) 从队列中取一个对象，如果给了block()，函数会一直阻塞到队列中有对象为止

18.9生产者-消费者问题

from random import randint
from time import sleep,ctime
from Queue import Queue
import threading
class MyThread(threading.Thread):
def __init__(self,func,args,name=''):
threading.Thread.__init__(self)
self.name=name
self.func=func
self.args=args
def getResult(self):
return self.res
def run(self):
print'starting', self.name,'at:',\
ctime()
self.res = apply(self.func,self.args)
print self.name, 'finished at:',\
ctime()
def writeQ(queue):
print 'producing object for Q...',
queue.put('xxx',1)
print "size now",queue.qsize()

def readQ(queue):
val = queue.get(1)
print 'consumed object fromQ... size now',\
queue.qsize()
def writer(queue,loops):
for i in range(loops):
writeQ(queue)
sleep(randint(1,3))

def reader(queue,loops):
for i in range(loops):
readQ(queue)
sleep(randint(2,5))

funcs = [writer,reader]
nfuncs = range(len(funcs))

def main():
nloops = randint(2,5)
q = Queue(32)
threads = []
for i in nfuncs:
t = MyThread(funcs[i],(q,nloops),funcs[i].__name__)
threads.append(t)
for i in nfuncs:
threads[i].start()

for i in nfuncs:
threads[i].join()

print 'all DONE'

if __name__ =='__main__':
main()

18.7练习

18-1 线程与进程的区别是什么？

进程：进程是程序的一次执行。每个进程都有自己的地址空间、内存、数据栈及其他记录其运行轨迹的辅助数据。

各个进程间使用进程间通讯（interprocess communication， IPC），而不能直接共享信息。

线程： 所有的线程运行在同一个进程中，共享相同的运行环境。

多进程的优点是：稳定性高，一个子进程崩溃了不会影响其他子进程和主进程（主进程死了，其他子进程都死），缺点是创建的代价大

多线程比多进程快一点，缺点是稳定性差，任何一个子线程的挂掉都会导致整个线程崩溃。

18-2. Python 的线程。在Python 中，哪一种多线程的程序表现得更好，I/O 密集型的还是计算密集型的？

I/O密集型。

计算密集型的特点是计算量大，消耗CPU资源，因此代码运行速率要求高，python脚本代码运行速率慢，不适合，适合C

I/O密集型涉及到网络，磁盘I/O任务的都是I/O密集型，CPU消耗少，适合开发效率高的语言，Python

18-3. 线程。你认为，多CPU 的系统与一般的系统有什么大的不同？多线程的程序在这种系统上的表现会怎么样？

实现真正的并发处理