Python自动化之线程进阶篇

拓展知识

什么是CPU-bound(计算密集型) 和I/O bound(I/O密集型) ？

I/O bound 指的是系统的CPU效能相对硬盘/内存的效能要好很多，此时，系统运作，大部分的状况是 CPU 在等 I/O (硬盘/内存) 的读/写，此时 CPU Loading 不高。
CPU bound 指的是系统的硬盘/内存效能相对 CPU 的效能要好很多，此时，系统运作，大部分的状况是 CPU Loading 100%，CPU 要读/写 I/O (硬盘/内存)，I/O在很短的时间就可以完成，而 CPU 还有许多运算要处理，CPU Loading 很高。
CPU bound的程序一般而言CPU占用率相当高。这可能是因为任务本身不太需要访问I/O设备，也可能是因为程序是多线程实现因此屏蔽掉了等待I/O的时间。
而I/O bound的程序一般在达到性能极限时，CPU占用率仍然较低。这可能是因为任务本身需要大量I/O操作，而pipeline做得不是很好，没有充分利用处理器能力

Python GIL(Global Interpreter Lock)

GIL是必要的，因为CPython的内存管理是非线程安全的。你不能简单地创建多个线程，并希望Python能在多核心的机器上运行得更快。这是因为 GIL將会防止多个原生线程同时执行Python字节码。换句话说，GIL將序列化您的所有线程。然而，您可以使用线程管理多个派生进程加速程序，这些程序独立的运行于你的Python代码外。

全局解释器锁GIL设计理念与限制

GIL的设计简化了CPython的实现，使得对象模型，包括关键的内建类型如字典，都是隐含可以并发访问的。锁住全局解释器使得比较容易的实现对多线程的支持，但也损失了多处理器主机的并行计算能力。

但是，不论标准的，还是第三方的扩展模块，都被设计成在进行密集计算任务是，释放GIL。

还有，就是在做I/O操作时，GIL总是会被释放。对所有面向I/O 的(会调用内建的操作系统C 代码的)程序来说，GIL 会在这个I/O 调用之前被释放，以允许其它的线程在这个线程等待I/O 的时候运行。如果是纯计算的程序，没有 I/O 操作，解释器会每隔 100 次操作就释放这把锁，让别的线程有机会执行（这个次数可以通过 sys.setcheckinterval 来调整）如果某线程并未使用很多I/O 操作，它会在自己的时间片内一直占用处理器（和GIL）。也就是说，I/O 密集型的Python 程序比计算密集型的程序更能充分利用多线程环境的好处。

threading模块

直接调用

import threading

import time

def sayhi(num): #定义每个线程要运行的函数

    print("running on number:%s" %num)

    time.sleep(3)

if __name__ == '__main__':

    t1 = threading.Thread(target=sayhi,args=(1,)) #生成一个线程实例

    t2 = threading.Thread(target=sayhi,args=(2,)) #生成另一个线程实例

    t1.start() #启动线程

    t2.start() #启动另一个线程

    print(t1.getName()) #获取线程名

    print(t2.getName())

间接调用

import threading

import time

class MyThread(threading.Thread):

    def __init__(self,num):

        threading.Thread.__init__(self)

        self.num = num

    def run(self):#定义每个线程要运行的函数

        print("running on number:%s" %self.num)

        time.sleep(3)

if __name__ == '__main__':

    t1 = MyThread(1)

    t2 = MyThread(2)

    t1.start()

    t2.start()

join(join()方法)和daemon（守护线程）

join方法实例

import threading

import time

class MyThread(threading.Thread):

        def __init__(self,id):

                threading.Thread.__init__(self)

                self.id = id

        def run(self):

                x = 0

                time.sleep(10)

                print self.id  

if __name__ == "__main__":

        t1=MyThread(999)

        t1.start()

        t1.join()

        for i in range(5):

                print i

t1.join()会等待子线程执行完毕之后，才执行父进程

daemon守护线程

import threading

import time

class MyThread(threading.Thread):

        def __init__(self,id):

                threading.Thread.__init__(self)

        def run(self):

                time.sleep(5)

                print "This is " + self.getName()  

if __name__ == "__main__":

        t1=MyThread(999)

        t1.setDaemon(True)

        t1.start()

        print "I am the father thread."

t1.setDaemon(True) 把t1线程设为守护线程，当主线程退出之后，不管主线程是否执行完都会退出。

线程锁(互斥锁Mutex)

import time

import threading

def addNum():

    global num #在每个线程中都获取这个全局变量

    print('--get num:',num )

    time.sleep(1)

    lock.acquire() #修改数据前加锁

    num  -=1 #对此公共变量进行-1操作

    lock.release() #修改后释放

num = 100  #设定一个共享变量

thread_list = []

lock = threading.Lock() #生成全局锁

for i in range(100):

    t = threading.Thread(target=addNum)

    t.start()

    thread_list.append(t)

for t in thread_list: #等待所有线程执行完毕

    t.join()

print('final num:', num )

RLock递归锁

锁里面包括锁

import threading,time

def run1(): # run1里面又给变量加了锁

    print("grab the first part data")

    lock.acquire()

    global num

    num +=1

    lock.release()

    return num

def run2():

    print("grab the second part data")

    lock.acquire()

    global  num2

    num2+=1

    lock.release()

    return num2

def run3():  #run3里面给run1和run2加了锁

    lock.acquire()

    res = run1()

    print('--------between run1 and run2-----')

    res2 = run2()

    lock.release()

    print(res,res2)

if __name__ == '__main__':

    num,num2 = 0,0

    lock = threading.RLock()

    for i in range(10):

        t = threading.Thread(target=run3)  # run3函数

        t.start()

while threading.active_count() != 1:

    print(threading.active_count())

else:

    print('----all threads done---')

    print(num,num2)

Semaphore(信号量)

信号量顾名思义就是控制线程运行的数量

import threading,time

def run(n):

    semaphore.acquire()

    time.sleep(1)

    print("run the thread: %s\n" %n)

    semaphore.release()

if __name__ == '__main__':

    num= 0

    semaphore  = threading.BoundedSemaphore(5) #最多允许5个线程同时运行

    for i in range(20):

        t = threading.Thread(target=run,args=(i,))

        t.start()

while threading.active_count() != 1:

    pass #print threading.active_count()

else:

    print('----all threads done---')

    print(num)

Events

线程间通信

import threading,time

import random

def light():

    if not event.isSet():

        event.set() #wait就不阻塞 #绿灯状态

    count = 0

    while True:

        if count < 10:

            print('\033[42;1m--green light on---\033[0m')

        elif count <13:

            print('\033[43;1m--yellow light on---\033[0m')

        elif count <20:

            if event.isSet():

                event.clear()

            print('\033[41;1m--red light on---\033[0m')

        else:

            count = 0

            event.set() #打开绿灯

        time.sleep(1)

        count +=1

def car(n):

    while 1:

        time.sleep(random.randrange(10))

        if  event.isSet(): #绿灯

            print("car [%s] is running.." % n)

        else:

            print("car [%s] is waiting for the red light.." %n)

if __name__ == '__main__':

    event = threading.Event()

    Light = threading.Thread(target=light)

    Light.start()

    for i in range(3):

        t = threading.Thread(target=car,args=(i,))

        t.start()

thareading.Event机制类似于一个线程向其它多个线程发号施令的模式，其它线程都会持有一个threading.Event的对象，这些线程都会等待这个事件的“发生”，如果此事件一直不发生，那么这些线程将会阻塞，直至事件的“发生”。

import threading

class mythread(threading.Thread):

    def __init__(self, threadname):

        threading.Thread.__init__(self, name=threadname)

    def run(self):

        global event

        if event.isSet():

            print("I am in if")

            event.clear()

            event.wait()

            print(self.getName())

        else:

            print("I am in else")

            print(self.getName())

            event.set()

event = threading.Event()

event.set()

t1 = []

for i in range(10):

    t = mythread(str(i))

    t1.append(t)

for i in t1:

    i.start()

event.wait()会阻塞直到有资源，也就是event.set()将标志设置为True。

巴特西