【Python与线程】

"

目录

一、全局解释器锁GIL

二、Python线程模块的选择

三、线程的创建

三、锁机制

四、信号量

五、事件

六、条件

七、定时器

八、线程队列

九、线程池

补充：线程安全

1. import threading
3. obj = threading.local()
4. # local():可实现，多线程操作某一数据，不会出现数据混乱的情况
5. # 原理：空间换时间
8. def add(i):
9. obj.n = i
10. print(i, obj.n, threading.current_thread().ident)
13. for i in range(20):
14. th = threading.Thread(target=add, args=(i,))
15. th.start()

 

一、全局解释器锁GIL

Python代码的执行由Python虚拟机(也叫解释器主循环)来控制。Python在设计之初就考虑到要在主循环中同时只有一个线程在执行。虽然Python解释器中可以“运行”多个线程，但在任意时刻只有一个线程在解释器中运行.

对于Python虚拟机的访问由全局解释器锁(GIL)来控制，正是这个锁保证了同一时刻只有一个线程在运行.

同一时间点，GIL只允许同一个进程中的一个线程访问cpu，即CPython解释器中没有真正的线程并行，只有进程可以实现。故I/O操作多时，使用多线程最好；计算密集时，使用多进程最好。

• 在多线程环境中，Python虚拟机按以下方式执行：

1. 设置GIL；

2. 切换到一个线程去执行；

3. 运行指定数量的字节码指令或者线程主动让出控制(如调用time.sleep(0))；

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

5. 解锁GIL；

6. 再次重复以上所有步骤；

在调用外部代码（如C/C++扩展函数）的时候，GIL将会被锁定，知道这个函数结束为止（由于在这期间没有Python的字节码被运行，所以不会做线程切换），编写扩展的程序员可以主动解锁GIL。

二、Python线程模块的选择

1. Python提供了多个用于多线程编程的模块，包括thread、threading和Queue等。thread和threading模块允许程序员创建和管理线程。thread模块提供了基本的线程和锁的支持；threading提供了更高级别、功能更强的线程管理功能。而Queue模块允许用户创建一个可以用于多个线程之间共享数据的队列数据结构。

2. 应避免使用thread模块，因为更高级别的threading模块更为先进，对线程的支持更加完善。而且thread模块里的属性有可能会与threading出现冲突；其次，低级别的thread模块的同步原理很少(实际上只有一个)，而threading模块则有很多；再则，thread模块中当主线程结束时，所有的子线程都会被强制结束掉，没有警告也不会做正常的清除工作，但threading模块至少能够确保重要的子线程退出后才退出主线程.

3.关于守护线程：thread模块不支持守护线程，当主线程退出时，所有的子线程不论它们是否还在工作，都会被强制退出。而threading模块支持守护线程，守护线程一般是一个等待客户请求的服务器，若没有客户提出请求它就在那等着，如果设定这个线程为守护线程，就表示这个线程是不重要的，在进程(主线程)退出的时候不用等待这个线程退出.

三、线程的创建

1. 创建线程

1. # 创建方式1
3. from threading import Thread
4. from time import sleep
7. def sayhi(name):
8. sleep(1)
9. print(name, 'say hello')
12. t = Thread(target=sayhi, args=('egon',))
13. t.start()
14. print("主线程")

1. # 创建方式2：自定义线程类
3. from threading import Thread
4. from time import sleep
7. class Sayhi(Thread):
9. def __init__(self, name):
10. super().__init__()
11. self.name = name
13. def run(self): # 必备方法，用于启动线程
14. sleep(1)
15. print(self.name, 'say hello')
18. t = Sayhi('egon')
19. t.start()
20. print("主线程")

2. 多线程与多进程

• pid比较

1. from threading import Thread
2. from multiprocessing import Process
3. from os import getpid
6. work = lambda who:print(who, getpid())
8. if __name__ == '__main__':
10. # part1: 在主进程下开启多个线程，每个线程都与主进程的pid一样
11. t1 = Thread(target=work, args=("线程",))
12. t2 = Thread(target=work, args=("线程",))
13. for t in (t1, t2):
14. t.start()
15. t.join()
17. # part2: 开启多个子进程，每个进程都有不同的pid
18. p1 = Process(target=work, args=("进程",))
19. p2 = Process(target=work, args=("进程",))
20. for p in (p1, p2):
21. p.start()
22. p.join()
24. print("主线程/主进程", getpid())

• 启动速度的较量

1. # 单线程与单进程
3. from threading import Thread
4. from multiprocessing import Process
7. work = lambda who:print("我是%s" % who)
9. if __name__ == '__main__':
11. p = Process(target=work, args=("进程",))
12. p.start()
14. t = Thread(target=work, args=("线程",))
15. t.start()
17. # 结果必然是线程先打印出来

1. from threading import Thread
2. from multiprocessing import Process
3. from time import time, sleep
6. def func():
7. pass
10. def work(Process, Thread):
11. # 注意：这是测试启动速度，所以无需join
13. # 进程
14. p_start = time()
15. [Process(target=func).start() for i in range(100)]
16. p_over = time() - p_start
18. # 线程
19. t_start = time()
20. [Thread(target=func).start() for i in range(100)]
21. t_over = time() - t_start
23. print("进程用时：%s\t\t线程用时：%s" % (p_over, t_over))
24. # 进程用时：0.21951699256896973 线程用时：0.015702009201049805
25. # 经过多次测试后得到结论：多线程的启动速度远快于多进程的启动速度
28. if __name__ == '__main__':
29. Process(target=work, args=(Process, Thread)).start()

3. 线程与进程内存数据共享的区别

1. from threading import Thread
2. from multiprocessing import Process
5. def work():
6. global n
7. n = 0
10. if __name__ == '__main__':
11. n = 1
13. p = Process(target=work)
14. p.start()
15. p.join()
16. print("父进程:", n)
17. # 显然，子进程中改的仅仅是它自己的全局n，不会影响到父进程
19. t = Thread(target=work)
20. t.start()
21. t.join()
22. print("主线程:", n)
23. # 此时查看结果为0，因为同一进程内的线程共享进程内的数据

4. 守护线程

无论是线程还是进程，都遵循守护线程(进程)等待主线程(进程)运行完毕后被销毁。需要强调的是，运行完毕并非终止运行：对于主进程来说，运行完毕指的是主进程代码运行完毕；对于主线程来说，运行完毕指的是主线程所在的进程内所有非守护线程统统运行完毕，主线程才算运行完毕.

• 详细说明

1. 主进程在其代码执行结束后就已经算运行完毕了（守护进程在此时就被回收），然后主进程会一直等非守护的子进程都运行完毕后回收子进程的资源（否则会产生僵尸进程），才会结束。

2. 主线程在其它非守护线程运行完毕后才算运行完毕（守护线程在此时就被回收），因为主线程的结束意味着进程的结束，进程整体的资源都将被回收，而进程必须保证非守护线程都运行完毕后才能结束。

• 实例

1. # 守护线程与守护进程的区别
3. from threading import Thread
4. from multiprocessing import Process
5. from time import sleep
8. def func(who, n):
9. sleep(n)
10. print("我是%s" % who)
13. if __name__ == '__main__':
14. # t1 = Thread(target=func, args=("守护线程", 1,))
15. # t2 = Thread(target=func, args=("普通线程", 2,))
16. # t1.setDaemon(True) # 必须在启动之前设置
17. # [t.start() for t in (t1, t2)]
18. # # 守护线程会在主进程内的所有普通线程运行完毕时立即终止
20. p1 = Process(target=func, args=("守护进程", 1))
21. p2 = Process(target=func, args=("普通进程", 2))
22. p1.daemon = True
23. [p.start() for p in (p1, p2)]
24. # 主进程结束，守护进程即终止
26. print("我是主线程")

5. Thread类的其它方法

• Thread实例对象的方法

isAlive()：返回线程是否活动

getName()：返回线程名

setName()：设置线程名

• threading模块提供的一些方法

  
  

currentThread()：返回当前的线程变量

enumerate()：返回正在运行的线程列表，不包括启动前和终止后的线程

activeCount()：返回正在运行的线程数量，等价于len(threading.enumerate()）

三、锁机制

1. 同步锁

1. # 多线程抢占资源的情况
3. from threading import Thread
4. from multiprocessing import Value
5. from time import sleep
8. def work():
9. global n
10. temp = n
11. sleep(0.1)
12. n = temp - 1
15. if __name__ == '__main__':
16. n = 100
17. t_lst = []
19. for i in range(100):
20. t = Thread(target=work)
21. t.start()
22. t_lst.append(t)
24. [t.join() for t in t_lst]
26. print(n) # 此时结果极有可能是99

1. # 同步锁的引用
3. from threading import Thread, Lock
6. def work():
7. global n
8. lock.acquire() # 加锁
9. temp = n
10. n = temp - 1
11. lock.release() # 释放
14. if __name__ == '__main__':
15. lock = Lock()
16. n = 100
17. t_lst = []
19. for i in range(100):
20. t = Thread(target=work)
21. t.start()
22. t_lst.append(t)
24. [t.join() for t in t_lst]
26. print(n) # 此时结果肯定为0
27. # 由原来的并发执行变成了串行执行，牺牲效率而保证了数据的安全性

• 关于互斥锁与join方法

在start之后立即用jion，也能使线程变为串行，保证数据安全。但问题是在start之后使用jion会使任务内的所有代码都变为串行执行，速度之慢可想而知。而加锁只会将任务内的部分代码变为串行执行。单从数据安全方面考虑，二者都可以实现，但很明显加锁的效率更高。

2. 死锁与递归锁

所谓死锁：是指两个或两个以上的进程(线程)在执行过程中，因争夺资源而造成的一种互相等待的现象，若无外力作用，它们都将无法推进下去。此时称系统处于死锁状态或系统产生了死锁，这些永远在等待的进程称为死锁进程，如下：

1. from threading import Lock
3. lock = Lock()
5. lock.acquire()
6. lock.acquire()
7. print(1)
8. lock.release()
9. lock.release()

解决方法：递归锁，在Python中为了支持同一个线程中多次请求同一资源，python提供了可重入锁RLock，RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源可以被多次require。直到一个线程内所有的acquire都被release，其它的线程才能获得资源，上面的例子如果使用RLock，则不会发生死锁：

1. from threading import RLock
3. rlock = RLock()
5. rlock.acquire()
6. rlock.acquire()
7. print(1)
8. rlock.release()
9. rlock.release()

• 典型问题：抢饭吃

1. # 陷入死锁
3. from threading import Thread, Lock
4. from time import sleep
7. noodle_lock = Lock()
8. fork_lock = Lock()
10. def eat1(name):
11. noodle_lock.acquire()
12. print("%s抢到了面" % name)
14. fork_lock.acquire()
15. print("%s抢到了叉子" % name)
17. fork_lock.release()
18. noodle_lock.release()
21. def eat2(name):
22. fork_lock.acquire()
23. print("%s抢到了叉子" % name)
24. sleep(1)
25. noodle_lock.acquire()
26. print("%s抢到了面" % name)
28. noodle_lock.release()
29. fork_lock.release()
32. for name in 'people1', 'people2':
33. Thread(target=eat1, args=(name,)).start()
34. Thread(target=eat2, args=(name,)).start()
36. # 此时是一个人抢到了面，一个人抢到了叉子，结果是都吃不上饭

1. # RLock解决死锁
3. from threading import Thread, RLock
6. noodle_lock = fork_lock = RLock()
8. def eat1(name):
9. noodle_lock.acquire()
10. print("%s抢到了面" % name)
12. fork_lock.acquire()
13. print("%s抢到了叉子" % name)
15. fork_lock.release()
16. noodle_lock.release()
19. def eat2(name):
20. fork_lock.acquire()
21. print("%s抢到了叉子" % name)
23. noodle_lock.acquire()
24. print("%s抢到了面" % name)
26. noodle_lock.release()
27. fork_lock.release()
30. for name in 'people1', 'people2':
31. Thread(target=eat1, args=(name,)).start()
32. Thread(target=eat2, args=(name,)).start()

四、信号量

Semaphore管理一个内置的计数器，每当调用acquire时，内置计数器-1；调用release时内置计数器+1；计数器不能小于0；当计数器为0时，acquire将阻塞线程，直到其它线程调用release.

信号量与进程池是完全不同的概念，进程池Pool(4)的意思是最大只能产生4个进程，而且从头到尾都只是这四个进程，不会产生新的，而信号量是产生一大堆线程/进程.

• 实例：（限制同时最多5个线程可以访问锁内资源）

1. from threading import Thread, Semaphore, current_thread
2. from time import sleep
5. def func():
6. s.acquire()
7. print('%s get sm' %current_thread().getName())
8. sleep(1)
9. s.release()
12. s = Semaphore(5) # 允许最多5个线程同时访问锁内资源
14. [Thread(target=func).start() for i in range(10)]

五、事件

• 介绍

线程的一个关键特性是每个线程都是独立运行且状态不可预测。如果程序中的其它线程需要通过判断某个线程的状态来确认自己下一步的操作，这时线程同步问题就会变得非常棘手。为了解决这些问题，我们需要使用threading库中的Event对象，此对象包含一个可由线程设置的信号标志，它允许线程等待某些事件的发生。在初始情况下，Event对象的信号标志被设置为假，如果有线程等待一个Event对象, 而这个Event对象的标志为假,那么这个线程将会被一直阻塞，直至该标志为真。一个线程如果将一个Event对象的信号标志设置为真,它将唤醒所有等待这个Event对象的线程。如果一个线程等待一个已经被设置为真的Event对象,那么它将忽略这个事件, 继续执行。

• 方法

event.isSet()：返回event的状态值，等价于event.is_set()

event.wait(timeout)：如果is_set的值为False，将阻塞线程，timeout是可选超时时间，默认永久

event.set()：设置event的状态值为True，所有阻塞池的线程激活进入就绪状态，等待操作系统调度

event.clear()：恢复event的状态值为False

例如，有多个工作线程尝试连接MySQL，我们想要在连接前确保MySQL服务正常才让那些工作线程去连接MySQL服务器，如果连接不成功，都会去尝试重新连接。那么我们就可以采用Event机制来协调各个工作线程的连接操作：

1. # 模拟连接MySQL
3. from threading import Thread, Event, current_thread
4. from time import sleep
5. from random import uniform
8. def conn_MySQL():
9. count = 1
10. while not event.is_set():
11. if count > 3:raise TimeoutError("连接超时")
12. print("<%s>第%s次尝试连接" %(current_thread().getName(), count))
13. event.wait(0.5) # 等待is_set的值为True，等待超时时间0.5秒
14. count +=1
15. print("<%s>连接成功" %current_thread().getName())
18. def check_MySQL():
19. print("\033[45m[%s]正在检查MySQL\033[0m" %current_thread().getName())
20. sleep(uniform(0, 2))
21. event.set() # 将is_set的值设为True
24. event = Event()
25. Thread(target=conn_MySQL).start()
26. Thread(target=conn_MySQL).start()
27. Thread(target=check_MySQL).start()

六、条件

• 介绍

Python提供的Condition对象提供了对复杂线程同步问题的支持。Condition被称为条件变量，除了提供与Lock类似的acquire和release方法外，还提供了wait和notify方法。线程首先acquire一个条件变量，然后判断一些条件。如果条件不满足则wait；如果条件满足，进行一些处理改变条件后，通过notify方法通知其他线程，其他处于wait状态的线程接到通知后会重新判断条件。不断的重复这一过程，从而解决复杂的同步问题。

• 实例

1. from threading import Thread, Condition
2. from time import sleep
4. def run(n):
5. c.acquire()
6. c.wait() # 阻塞，等待notify发送信号，wait方法可指定超时时间，默认永久阻塞，
7. # 实测：若超时时间到还未收到信号则直接执行下面的代码
8. print('run the thread:', n)
9. c.release()
12. c = Condition()
13. [Thread(target=run, args=(i,)).start() for i in range(10)]
15. while 1:
16. sleep(0.1)
17. num = int(input('>>>'))
18. if num == 0:break
19. c.acquire()
20. c.notify(num) # 给wait发信号, num指定发多少个信号
21. c.release()

七、定时器

指定n秒后执行某个操作

1. from threading import Timer
3. func = lambda :print('hello, world')
4. Timer(3, func).start() # 指定3秒后执行func

八、线程队列

queue is especially useful in threaded programming when information must be exchanged safely between multiple threads.

译文：当必须在多个线程之间安全地交换信息时，队列在线程编程中特别有用。

• 先进先出与先进后出

1. # 先进先出
3. from queue import Queue
5. q = Queue(3) # 实例化一个Queue对象，指定队列最大可放3个值，同理LifoQueue，和优先级的队列
6. q.put('first') # 往队列中放值
7. q.put('second')
8. q.put('third')
9. # q.put('return') # 若超过队列指定的最大值，将阻塞等待其它线程取值，同理lq.put，和优先级的队列
11. print(q.get()) # 从队列中取值
12. print(q.get())
13. print(q.get())
15. ############################
17. # 先进后出
19. from queue import LifoQueue
21. lq = LifoQueue()
22. lq.put('first') # 往队列中放值
23. lq.put('second')
24. lq.put('third')
26. print(lq.get()) # 从队列中取值
27. print(lq.get())
28. print(lq.get())

• 优先级的队列

1. from queue import PriorityQueue
3. q = PriorityQueue(3) # 可指定队列
5. # put(tuple)，tuple的第一个元素是优先级，第二个元素是要放入的值
6. # (通常是数字，也可以是非数字之间的比较，数字越小优先级额越高)
7. # 优先级队列比较的ascii码
8. q.put((2, 'b'))
9. q.put((1, 'a'))
10. q.put((3, 'c'))
12. for i in range(3):print(q.get())
13. """
14. 结果（数字越小优先级越高，优先级高的优先出队）
15. (1, 'a')
16. (2, 'b')
17. (3, 'c')
18. """

• 更多方法说明

优先队列的构造函数。maxsize是一个整数，它设置可放在队列中的项数的上限限制。一旦达到此大小，插入将阻塞，直到使用队列项为止。如果maxsize小于或等于0，则队列大小为无穷大。

首先检索值最低的条目(值最低的条目是通过排序(list(entries))[0]返回的条目)。条目的典型模式是表单中的元组:(priority_number, data)。

异常queue.Empty

当对空的队列对象调用非阻塞get()(或get_nowait())时引发异常。

异常queue.Full

当对已满的队列对象调用非阻塞put()(或put_nowait())时引发的异常。

Queue.qsize()

如果为空，则返回True

如果已满，则返回True

队列中。把(项目,块= True,超时= None)

将项目放入队列中。如果可选的args块为true，超时为None(默认)，则在空闲插槽可用之前，如果有必要，阻塞。如果超时是正数，它会阻塞大多数超时秒，如果在此时间内没有可用的空闲时间，则会引发完全异常。否则(block为false)，如果一个空闲的插槽立即可用，就将一个条目放到队列中，否则引发完整的异常(在这种情况下会忽略超时)。

Queue.put_nowait(项)

相当于把(项目,假)。

队列中。得到(块= True,超时= None)

从队列中删除并返回项。如果可选的args块为true，而超时为None(默认值)，则在项可用之前，如果有必要，阻塞。如果超时是正数，它会阻塞大多数超时秒，如果在这段时间内没有可用项，就会引发空异常。否则(block为false)，如果一个项立即可用，返回一个项，否则引发空异常(在这种情况下忽略超时)。

Queue.get_nowait()

相当于(False)。

提供了两种方法来支持跟踪入队列任务是否由守护进程使用者线程完全处理。

Queue.task_done()

指示以前加入队列的任务已经完成。由队列使用者线程使用。对于用于获取任务的每个get()，对task_done()的后续调用将告诉队列任务的处理已经完成。

如果一个join()当前处于阻塞状态，那么当所有项都被处理完时(这意味着对于每个已将()放入队列的项都接收了task_done()调用)，它将恢复。

如果调用的次数超过了队列中放置的项的次数，就会引发ValueError。

九、线程池

• Python标准模块concurrent.futures提供了高度封装的异步调用接口:

ThreadPoolExecutor：线程池，提供异步调用

ProcessPoolExecutor：进程池，提供异步调用

• 基本方法：

submit(fn, *args, **kwargs)：异步提交任务

map(func, *iterables, timeout=None, chunksize=1)：取代for循环submit的操作

shutdown(wait=True)：相当于Pool的close+join操作。wait=True：等待池内所有任务执行完毕回收完资源后才继续；wait=false：立即返回，并不会等待池内的任务执行完毕；但不管wait参数为何值，整个程序都会等到所有任务执行完毕，submit或map必须写在shutdown之前。

add_done_callback(fn)：回调函数，线程属于子线程的调用，进程属于父进程的调用

result(timeout=None)：取得结果

• 实例一：回调函数

1. # 回调函数的用法
3. from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
4. from os import cpu_count # 用于获取cpu核心数
7. def func(i):
8. return i * i
11. # 用于回调函数
12. def func2(i):
13. print(i.result() ** 0.5)
16. # 错误:
17. # func = lambda i:i*i
18. # func2 = lambda i:print(i.result() ** 2)
19. # 实测ProcessPoolExecutor池无法调用匿名函数，包括Pool池
22. ####################################
23. # 进程
25. # p = ProcessPoolExecutor((cpu_count() or 1) + 1)
26. # for i in range(10):
27. # p.submit(func, i).add_done_callback(func2)
28. # p.shutdown() # 等价于Pool的close+join
31. ####################################
32. # 线程
34. t = ThreadPoolExecutor(10)
35. for i in range(10):
36. t.submit(func, i).add_done_callback(func2)
39. # 线程是串行，所以异步执行的打印结果是有序的，而进程是并发/并行，故无须

• 实例二：计算型速度对比

1. # 计算型速度对比
3. from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
4. from multiprocessing import Pool
5. from os import cpu_count
6. from time import time
9. def func(i):
10. num = 0
11. for v in range(i):
12. num += v ** v
15. # ThreadPoolExecutor
16. start = time()
17. t = ThreadPoolExecutor(1)
18. t.map(func, range(1000))
19. t.shutdown()
20. print("线程池用时：", time() - start)
21. # 线程池用时： 4.917341947555542
24. # ProcessPoolExecutor
25. start = time()
26. p = ProcessPoolExecutor((cpu_count() or 1 ) + 1)
27. # [p.submit(func,i) for i in range(1000)]
28. p.map(func, range(1000))
29. p.shutdown()
30. print("进程池用时：", time() - start)
31. # 进程池用时： 2.890726089477539
34. # Pool异步
35. start = time()
36. pool = Pool((cpu_count() or 1) + 1)
37. # [pool.apply_async(func, args=(i,)) for i in range(1000)]
38. # pool.close()
39. # pool.join()
40. pool.map(func, range(1000))
41. print("进程池(Pool)用时：", time() - start)
42. # 进程池(Pool)用时： 2.6030900478363037