python多线程之二——threading模块

上一篇文章讲了python多线程的基础知识和thread模块，这一篇着重讲解一下threading模块

threading模块

threading模块除了Thread类之外，好包括其他很多的同步机制，下面来看一下threading模块汇总所包含的对象。

对象	描述
Thread	执行线程的对象
Lock	锁对象
RLock	递归锁，是一个线程可以再次拥有已持有的锁对象
Condition	条件变量对象，使一个线程等待另一个线程满足特定的条件触发
Event	事件对象，普通版的Condition
Semaphore	信号量，为线程间共享的资源提供一个“计数器”，计数开始值为设置的值，默认为1
BoundedSemaphore	与Semaphore相同，有边界，不能超过设置的值
Timer	定时运行的线程对象，定时器
Barrier	界限，当达到某一界限后才可以继续执行

看到threading有这么多对象，是不是有些懵了，下面一个个的来看一下

Thread类

Thread类是threading模块的主要主要执行对象。Thread对象有三个数据属性，name(线程名)、ident(线程的标识)、daemon(布尔值，是否是守护线程)。这三个数据属性可以直接通过对象进行调用并进行设置。

守护线程一般是一个等待客户端请求服务的服务器，进程退出时，该线程在正常情况下不会退出
Thread类还有一些对象方法

对象方法	描述
__init__()	实例化一个线程对象
start()	开始执行线程
run()	定义线程功能方法(一般在子类中进行重写)
join(timeout=None)	直至启动的线程终止或timeout秒，否则一直挂起，多用于主线程进行阻塞等待子线程运行完毕。
isAlivel/is_alive()	线程是否存活

注意
__init__()完整函数如下__init__(group=None,target=None,name=None,args=(),kwargs={},verbose=None,daemon=None)，Thread对象实例化需要一个可调用的target(可以是一个函数，也可是一个可调用的类实例)，参数args或者kwargs。

说了这么多，那怎么创建线程呢？一般有两种方法：

• 创建Thread实例，传给其一个函数或可调用的类实例
• 派生Thread的子类，并创建子类的实例
  一般来说，创建Thread实例并传递一个函数和派生Thread子类比较常用，后者更符合面向对象且比较容易扩展

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

import random
import threading
from time import ctime,sleep

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

def main():
    print('starting at:',ctime())
    threads = []

    for i in range(3):
        t = threading.Thread(target=loop,args=(i,random.randint(1,5)))
        threads.append(t)

    for i in range(3):
        threads[i].start()

    for i in range(3):
        threads[i].join()

    print('all done at:',ctime())

if __name__ == '__main__':
    main()

传递给Thread实例一个函数其实和thread模块中差不多，这里随机生成3个Thread实例，分别运行随机事件，然后通过循环让线程启动threads[i].start()，然后通过join()让主线程等待结束。
打印结果如下

starting at: Thu Sep  7 17:53:44 2017
start loop  0  sec: 1  at: Thu Sep  7 17:53:44 2017
start loop  1  sec: 5  at: Thu Sep  7 17:53:44 2017
start loop  2  sec: 3  at: Thu Sep  7 17:53:44 2017
end loop  0  done at: Thu Sep  7 17:53:45 2017
end loop  2  done at: Thu Sep  7 17:53:47 2017
end loop  1  done at: Thu Sep  7 17:53:49 2017
all done at: Thu Sep  7 17:53:49 2017

如果将join代码注释掉的话，主线程将不会等待子线程运行，打印结果如下:

starting at: Thu Sep  7 17:56:11 2017
start loop  0  sec: 2  at: Thu Sep  7 17:56:11 2017
start loop  1  sec: 4  at: Thu Sep  7 17:56:11 2017
start loop  2  sec: 5  at: Thu Sep  7 17:56:11 2017
all done at: Thu Sep  7 17:56:11 2017
end loop  0  done at: Thu Sep  7 17:56:13 2017
end loop  1  done at: Thu Sep  7 17:56:15 2017
end loop  2  done at: Thu Sep  7 17:56:16 2017

可以看到all done语句已经先执行完毕。然后各个子线程仍然在运行直到结束。

2. 创建Thread实例，传给它一个可调用类的实例
   这里需要解释一下，一般实现了__call__方法的类，其实例可以像函数一样进行调用(其实函数就是可调用的对象)，称之为可调用类的实例。这样的话，只需要在类中实现__call__方法即可

import random
import threading
from time import ctime, sleep

class ThreadFunc(object):
    def __init__(self,func,args,name=''):
        self.name = name
        self.func = func
        self.args = args

    def __call__(self):
        self.func(*self.args)

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

def main():
    print('start at:',ctime())
    threads = []
    loops = range(3)
    for i in loops:
        t = threading.Thread(target=ThreadFunc(loop,(i,random.randint(1,5)),loop.__name__))
        threads.append(t)

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

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

    print('all done at:',ctime())

if __name__ == '__main__':
    main()

这个例子和上面的例子相同，只不过在实例化Thread的时候将ThreadFunc传递给target，当t调用start的时候，其会调用__call__方法。
看一下运行的结果：

import random
import threading
from time import ctime, sleep


class ThreadFunc(object):
    def __init__(self,func,args,name=''):
        self.name = name
        self.func = func
        self.args = args

    def __call__(self):
        self.func(*self.args)

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

def main():
    print('start at:',ctime())
    threads = []
    loops = range(3)
    for i in loops:
        t = threading.Thread(target=ThreadFunc(loop,(i,random.randint(1,5)),loop.__name__))
        threads.append(t)

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

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

    print('all done at:',ctime())

if __name__ == '__main__':
    main()

3. 派生Thread的子类，创建子类的实例。
   派生Thread的子类，一般需要重写run方法。

import random
from threading import Thread
from atexit import register
from time import ctime, sleep

class ThreadFunc(Thread):
    def __init__(self,func,args):
        Thread.__init__(self)
        self.func = func
        self.args = args
        
    def run(self):
        self.func(*self.args)
        
def loop(nloop,nsec):
    print('start loop ',nloop,' sec:',nsec,' at:',ctime())
    sleep(nsec)
    print('end loop ',nloop,' at:',ctime())

def main():
    print('start at:',ctime())
    threads = []
    loops = range(3)
    for i in loops:
        t = ThreadFunc(loop,(i,random.randint(1,5)))
        threads.append(t)

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

@register
def _atexit():
    print('all done at:',ctime())

if __name__ == '__main__':
    main()

需要注意的是，子类的构造函数必须先调用基类的构造函数，在基类的构造函数中对于相应的参数进行了设置。这里并没有用join来控制主线程等待子线程完成，而是使用atexit.register()来注册一个退出函数。来看一下结果：

start at: Thu Sep  7 18:33:22 2017
start loop  0  sec: 4  at: Thu Sep  7 18:33:22 2017
start loop  1  sec: 3  at: Thu Sep  7 18:33:22 2017
start loop  2  sec: 4  at: Thu Sep  7 18:33:22 2017
end loop  1  at: Thu Sep  7 18:33:25 2017
end loop  0  at: Thu Sep  7 18:33:26 2017
end loop  2  at: Thu Sep  7 18:33:26 2017
all done at: Thu Sep  7 18:33:26 2017

以上就是多线程的三种实现方式。

同步原语

一般在多线程代码中，一般有一些特定的函数或代码块不希望被多个线程同时执行，这就需要使用同步了。同步原语中，锁是最简单、最低级的机制，而信号量通常用于多线程竞争有限资源的情况。

锁（Lock）

锁只有两种状态，锁定和解锁。即支持两种操作：获得锁和释放锁。当多线程争夺锁的时候，允许第一个获得锁的线程进入临界区，并执行代码，其他后到达的线程将被阻塞，直到获得锁的线程执行完毕，退出临界区，并释放锁。其他等待的线程去争夺锁并进行临界区。
下面来看一下不适用锁的例子。

import random
from atexit import register
from threading import currentThread, Thread
from time import ctime, sleep

remaining = []

def loop(nsec):
    tname = currentThread().name
    remaining.append(tname)
    print('{0} started at:{1}'.format(tname,ctime()))
    sleep(nsec)
    remaining.pop()
    print('{0} Completed at:{1} used {2}'.format(tname,ctime(),nsec))
    print(remaining)

def main():
    for i in range(4):
        Thread(target=loop,args=(random.randint(1,5),)).start()

@register
def _axexit():
    print('all down at:',ctime())

if __name__ == '__main__':
    main()

上面的程序很简单，一个共享的列表remaining来存储剩余的线程，在每个线程中先将线程名称添加到该列表，运行完成后，则从列表中删除。下面来看一下运行的结果:

Thread-1 started at:Thu Sep  7 22:11:01 2017
Thread-2 started at:Thu Sep  7 22:11:01 2017
Thread-3 started at:Thu Sep  7 22:11:01 2017
Thread-4 started at:Thu Sep  7 22:11:01 2017
Thread-2 Completed at:Thu Sep  7 22:11:02 2017 used time 1
Thread-3 Completed at:Thu Sep  7 22:11:02 2017 used time 1
['Thread-1', 'Thread-2']
['Thread-1', 'Thread-2']
Thread-4 Completed at:Thu Sep  7 22:11:03 2017 used time 2
['Thread-1']
Thread-1 Completed at:Thu Sep  7 22:11:04 2017 used time 3
[]
all down at: Thu Sep  7 22:11:04 2017

看上面的输出，Thread-2已经执行完毕，下面的remainlist却仍还有Thread-2，而Thread-4正在执行，却没有。这说明多个线程并行执行IO程序，同时有多个程序修改同一个变量导致值输出问题。这个时候需要用锁来防止多个线程同时修改共享数据。
修改代码如下:

import random
from atexit import register
from threading import currentThread, Thread, Lock
from time import ctime, sleep

remaining = []
lock = Lock()
def loop(nsec):
    tname = currentThread().name
    lock.acquire()
    remaining.append(tname)
    print('{0} started at:{1}'.format(tname,ctime()))
    lock.release()
    sleep(nsec)
    lock.acquire()
    remaining.pop()
    print('{0} Completed at:{1} used time {2}'.format(tname,ctime(),nsec))
    print(remaining)
    lock.release()

def main():
    for i in range(4):
        Thread(target=loop,args=(random.randint(1,5),)).start()

@register
def _axexit():
    print('all down at:',ctime())

if __name__ == '__main__':
    main()

大部分代码都没有改动，只是在添加了全局变量lock，然后在remaining添加和删除线程名称的时候进行锁的获取和释放，来看一下运行的结果。

Thread-1 started at:Thu Sep  7 22:26:40 2017
Thread-2 started at:Thu Sep  7 22:26:40 2017
Thread-3 started at:Thu Sep  7 22:26:40 2017
Thread-4 started at:Thu Sep  7 22:26:40 2017
Thread-1 Completed at:Thu Sep  7 22:26:41 2017 used time 1
['Thread-1', 'Thread-2', 'Thread-3']
Thread-4 Completed at:Thu Sep  7 22:26:42 2017 used time 2
['Thread-1', 'Thread-2']
Thread-2 Completed at:Thu Sep  7 22:26:42 2017 used time 2
['Thread-1']
Thread-3 Completed at:Thu Sep  7 22:26:44 2017 used time 4
[]
all down at: Thu Sep  7 22:26:44 2017

这样看结果是不是正常了。
可能你会觉得lock这样进行锁的获取和释放代码不太直观，你也可以使用with语句，如下：

with lock:
    remaining.append(tname)
    print('{0} started at:{1}'.format(tname,ctime()))

这样看起来是不是简洁多了，缩进的代码会自动进行加锁和释放锁的功能。

信号量

信号量其实是一个计数器，当资源消耗时递减，当资源释放时递增。资源的消耗与释放称为一次PV操作，P()源于probeer/proberen，也称为wait、try、acquire、pend或procure，是消耗资源使计数器递减的操作。V()源于verhogen/verhoog，也成为signal、increment、release、post、vacate，是释放资源，使其回到资源池的操作。不过在python中被固定为acquire和release操作。

下面通过生产者-消费者问题来了解一下信号量
场景：生产者生产产品，消费者消费产品，柜台上最多只能摆放5件产品

from atexit import register
from random import randrange
from threading import Lock, Thread, BoundedSemaphore
from time import sleep, ctime

lock = Lock()
MAX = 5
product_on_shelf = BoundedSemaphore(MAX)

def product():
    with lock:
        print('product is doing')
        try:
            product_on_shelf.release()
        except ValueError:
            print('shelf is full')
        else:
            print('product is done')

def consume():
    with lock:
        print('product is consume')
        if product_on_shelf.acquire():
            print('product is successed consume')
        else:
            print('shelf is empty')

def producer(loops):
    print('producer is start for loops:',loops)
    for i in range(loops):
        product()
        sleep(randrange(3))

def customer(loops):
    print('customer is start for loops:', loops)
    for i in range(loops):
        consume()
        sleep(randrange(3))

def main():
    print('starting at:',ctime())
    nloops = randrange(3,6)
    Thread(target=producer,args=(nloops,)).start()
    Thread(target=customer,args=(nloops,)).start()

@register
def _atexit():
    print('all done at:',ctime())

if __name__ == '__main__':
    main()

这里定义了生产者和消费者两个线程，随机循环生产和消费产品。product和consume是生产产品和消费产品的函数，均通过锁进行同步，这里使用的是BoundedSemaphore，主要是因为其有边界，值为Max，初始状态下信号量的值为5。这里需要对release进行异常捕获，超过边界BoundedSemaphore会抛出ValueError异常，而acquire则返回boolean来表示获取是否成功。下面来看一下结果：

starting at: Fri Sep  8 11:34:46 2017
producer is start for loops: 3
product is doing
shelf is full
customer is start for loops: 3
product is consume
product is successed consume
product is consume
product is successed consume
product is doing
product is done
product is consume
product is successed consume
product is doing
product is done
all done at: Fri Sep  8 11:34:50 2017

可以看到运行结果正常。

threading模块是对thread模块的封装，并添加了状态管理，并且通过lock扩展出来的Condition进行了多线程的通过管理，所以一般在实际生产中主要使用threading模块去处理问题。
threading模块暂时就讲到这里。