Python小白到老司机，快跟我上车！基础篇(十九)
Posted coder-pig
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Python小白到老司机，快跟我上车！基础篇(十九)相关的知识，希望对你有一定的参考价值。
1、threading模块详解
Python提供的与线程操作相关的模块，网上有很多资料还是用的thread模块，在3.x版本中已经使用
threading来替代thread，如果你在python 2.x版本想使用threading的话，可以使用dummy_threading模块。
① threading模块提供的可直接调用的函数
- active_count()：获取当前活跃(alive)线程的个数。
- current_thread()：获取当前的线程对象。
- get_ident()：返回当前线程的索引，一个非零的整数(3.3新增)。
- enumerate()：获取当前所有活跃线程的列表。
- main_thread()：返回主线程对象(3.4新增)。
- settrace(func)：设置一个回调函数，在run()执行之前被调用。
- setprofile(func)：设置一个回调函数，在run()执行完毕之后调用。
- stack_size()：返回创建新线程时使用的线程堆栈大小。
- threading.TIMEOUT_MAX：堵塞线程时间最大值，超过这个值会栈溢出。
② 线程局部变量(Thread-Local Data)
问题引入：
在一个进程内所有的线程共享进程的全局变量，线程间共享数据很方便但是每个线程都可以随意修改全局变量，可能会引起线程安全问题。
解决方法：
对于这种线程私有数据，最简单的方法就是对变量加锁或使用局部变量，只有线程自身可以访问，其他线程无法访问。除此之外还可以使用threading模块为我们提供的
ThreadLocal变量，它本身是一个全局变量，但是线程们却可以使用它来保存私有数据。
用法简介：
定义一个全局变量：data = thread.local()，然后就可以往里面存数据啦，比如data.num = xxx，但是有一点要注意：如果data里没有设置对应的属性，直接取会报AttributeError异常，使用时可以捕获这个异常或先调用hasattr(对象，属性)判断对象中是否有该属性！使用代码示例如下：
import threading
import random
data = threading.local()
def show(d):
try:
num = d.num
except AttributeError:
print("线程 %s 还未设置该属性！" % threading.current_thread().getName())
else:
print("线程 %s 中该属性的值为 = %s" % (threading.current_thread().getName(), num))
def thread_call(d):
show(d)
d.num = random.randint(1, 100)
show(d)
if __name__ == '__main__':
show(data)
data.num = 666
show(data)
for i in range(2):
t = threading.Thread(target=thread_call, args=(data,), name="Thread " + str(i))
t.start()
运行结果如下：
线程 MainThread 还未设置该属性！
线程 MainThread 中该属性的值为 = 666
线程 Thread 0 还未设置该属性！
线程 Thread 0 中该属性的值为 = 80
线程 Thread 1 还未设置该属性！
线程 Thread 1 中该属性的值为 = 17
不同线程访问这个ThreadLocal变量，返回的都是不一样的值，原理：
threading.local()实例化一个全局对象，这个全局对象里有一个大字典，键值为两个弱引用对象{线程对象，字典对象}，然后可以通过
current_thread()获得当前的线程对象，然后根据这个对象可以拿到对应的字典对象，然后进行参数的读或者写。
③ 线程对象(threading.Thread)
创建新线程的两种方式：
- 1.直接创建threading.Thread对象，并把调用对象作为参数传入；
- 2.继承threading.Thread类，重写run()方法；
使用代码示例（验证单线程快还是多线程快）：
import threading
import time
def catch_fish():
pass
def one_thread():
start_time = time.time()
for i in range(1, 1001):
catch_fish()
end_time = time.time()
print("单线程测试 耗时 === %s" % str(end_time - start_time))
def muti_thread():
start_time = time.time()
for i in range(1, 1001):
threading.Thread(target=catch_fish()).start()
end_time = time.time()
print("多线程测试 耗时 === %s" % str(end_time - start_time))
if __name__ == '__main__':
# 单线程
threading.Thread(one_thread()).start()
# 多线程
muti_thread()
运行结果如下：
单线程测试 耗时 === 0.00011301040649414062
多线程测试 耗时 === 0.07665514945983887
从输出结果可以看到，多线程反而比单线程要慢，原因是前面介绍过的Python中的全局解释器锁（GIL）， 使得任何时候仅有一个线程在执行。
Thread类构造函数
def __init__(self, group=None, target=None, name=None,
args=(), kwargs=None, *, daemon=None):
构造函数参数依次是：
- group：线程组
- target：要执行的函数
- name：线程名字
- args/kwargs：要传入的函数的参数
- daemon：是否为守护线程
相关属性与函数：
- start()：启动线程，只能调用一次
- run()：线程执行的操作，可继承Thread重写，参数可从args和kwargs获取；
- join([timeout])：堵塞调用线程，直到被调用线程运行结束或超时；如果
没设置超时时间会一直堵塞到被调用线程结束。- name/getName()：获得线程名；
- setName()：设置线程名；
- ident：线程是已经启动，未启动会返回一个非零整数；
- is_alive()：判断是否在运行，启动后，终止前；
- daemon/isDaemon()：线程是否为守护线程；
- setDaemon()：设置线程为守护线程；
④ Lock(指令锁)与RLock(可重入锁)
在概念那里就讲了，多个进程并发的访问临界资源可能会引起线程同步安全问题，写个简单的例子，然后再引入同步锁。代码示例如下：
import threading
file_name = "test.txt"
# 定义一个写入文件的方法
def write_to_file(msg):
try:
with open(file_name, "a+", encoding="utf-8") as f:
f.write(msg + "\\n")
except OSError as reason:
print(str(reason))
class MyThread(threading.Thread):
def __init__(self, msg):
super().__init__()
self.msg = msg
def run(self):
write_to_file(self.name + "~" + self.msg)
if __name__ == '__main__':
for i in range(1, 21):
t = MyThread(str(i)).start()
运行结果如下：
# test.txt文件内容
Thread-1~1
Thread-5~5
Thread-3~3
Thread-2~2
Thread-4~4
Thread-6~6
Thread-7~7
Thread-8~8
Thread-10~10
Thread-9~9
Thread-11~11
Thread-13~13
Thread-12~12
Thread-14~14
Thread-15~15
Thread-16~16
Thread-17~17
Thread-19~19
Thread-20~20
Thread-18~18
发现结果并没有按照我们预想的1-20那样顺序打印，而是乱的，threading模块中提供了两个类来确保多线程共享资源的访问：「Lock」 和 「RLock」。
Lock：指令锁，有两种状态(锁定与非锁定)，以及两个基本函数：
使用
acquire()设置为locked状态，使用release()设置为unlocked状态。acquire()函数有两个可选参数：blocking=True：是否堵塞当前线程等待；timeout=None：堵塞等待时间。如果成功获得lock，acquire返回True，否则返回False，超时也是返回False。使用起来也很简单，在访问共享资源的地方acquire一下，用完release就好。使用代码示例如下：
import threading
file_name = "test.txt"
lock = threading.Lock()
# 定义一个写入文件的方法(加锁)
def write_to_file(msg):
if lock.acquire():
try:
with open(file_name, "a+", encoding="utf-8") as f:
f.write(msg + "\\n")
except OSError as reason:
print(str(reason))
finally:
lock.release()
class MyThread(threading.Thread):
def __init__(self, msg):
super().__init__()
self.msg = msg
def run(self):
write_to_file(self.name + "~" + self.msg)
if __name__ == '__main__':
for i in range(1, 101):
t = MyThread(str(i)).start()
这里把循环次数改成了101，反复执行多次，test.txt中写入顺序也是正确的，加锁有效。另外有一点要注意：如果锁的状态是unlocked，此时调用release会抛出RuntimeError异常！
RLock：可重入锁，和Lock类似，但RLock却可以被同一个线程请求多次！ 举个例子：在一个线程里调用Lock对象的acquire方法两次。
lock.acquire()
lock.acquire()
lock.release()
lock.release()
你会发现程序卡住不动，因为已经发生了死锁，但是方法调用是在同一个线程里的，这很不合理吧。这个时候就可以引入RLock了，使用RLock编写一样代码，只需把threading.Lock()改成threading.RLock()，即可解决这个问题。
虽然使用RLock可以规避同一个线程引起的死锁问题，但是acquire和release函数要成对出现，即有多少个acquire就要有多少个release，才能够正真释放锁。
⑤ 条件变量(Condition)
上面的互斥锁Lock和RLock只是最简单的同步机制，Python为我们提供了Condition(条件变量)，以便于处理复杂线程同步问题，比如最经典的生产者与消费者问题。Condition除了提供与Lock类似的 acquire() 与 release() 函数外，还提供了 wait() 与 notify() 函数。
用法简介：
- 1.调用
threading.Condition获得一个条件变量对象；- 2.线程调用acquire获得Condition对象；
- 3.进行条件判断，不满足条件调用wait函数，满足条件，进行一些处理改变条件后，调用notify函数通知处于wait状态的线程，重新进行条件判断。
代码示例如下（实现一个简单的消费者和生产者）：
import threading
import time
condition = threading.Condition()
products = 0 # 商品数量
# 定义生产者线程类
class Producer(threading.Thread):
def run(self):
global products
while True:
if condition.acquire():
if products >= 99:
condition.wait()
else:
products += 2
print(self.name + "生产了2个产品，当前剩余产品数为：" + str(products))
condition.notify()
condition.release()
time.sleep(2)
# 定义消费者线程类
class Consumer(threading.Thread):
def run(self):
global products
while True:
if condition.acquire():
if products < 4:
condition.wait()
else:
products -= 4
print(self.name + "消耗了4个产品，当前剩余产品数为：" + str(products))
condition.notify()
condition.release()
time.sleep(2)
if __name__ == '__main__':
# 创建五个生产者线程
for i in range(5):
p = Producer()
p.start()
# 创建两个消费者线程
for j in range(2):
c = Consumer()
c.start()
部分运行结果如下：
Thread-1生产了2个产品，当前剩余产品数为：2
Thread-2生产了2个产品，当前剩余产品数为：4
Thread-3生产了2个产品，当前剩余产品数为：6
Thread-4生产了2个产品，当前剩余产品数为：8
Thread-5生产了2个产品，当前剩余产品数为：10
Thread-6消耗了4个产品，当前剩余产品数为：6
Thread-7消耗了4个产品，当前剩余产品数为：2
Thread-1生产了2个产品，当前剩余产品数为：4
Thread-5生产了2个产品，当前剩余产品数为：6
Thread-3生产了2个产品，当前剩余产品数为：8
Thread-7消耗了4个产品，当前剩余产品数为：4
Thread-6消耗了4个产品，当前剩余产品数为：0
Thread-4生产了2个产品，当前剩余产品数为：2
Condition维护着一个互斥锁对象(默认是RLock)，也可以自己实例化一个在Condition实例化的时候通过构造函数传入，所以，调用的Condition的acquire与release函数，其实调用就是这个锁对象的acquire与release函数。
Condition提供的其他函数：
- wait(timeout=None)：释放锁，同时线程被挂起，直到收到通知被唤醒
或超时(如果设置了timeout)，当线程被唤醒并重新占有锁时，程序才继续执行；- wait_for(predicate, timeout=None)：等待知道条件为True，predicate应该是
一个回调函数，返回布尔值，timeout用于指定超时时间，返回值为回调函数返回
的布尔值，或者超时，返回False(3.2新增)；- notify(n=1)：默认唤醒一个正在的等待线程，notify并不释放锁！！！
- notify_all()：唤醒所有等待线程，进入就绪状态，等待获得锁，notify_all 同样不释放锁！！！
注：上述函数只有在acquire之后才能调用，不然会报RuntimeError异常。
⑥ 信号量(Semaphore)
信号量，也是一个简单易懂的东西，举个形象的例子：
厕所里有五个坑位，每有个人去厕所就会占用一个坑位，所剩余的坑位-1，当五个坑都被人占满时，新来的人就只能在外面等候，直到有人出来为止。这里的五个坑位就是信号量，蹲坑的人就是线程，初始值为5，来人-1，走人+1，超过最大值，新来的处于堵塞状态，我们写下代码来还原这个过程。
信号量使用代码示例如下：
import threading
import time
import random
s = threading.Semaphore(5) # 粪坑
class Human(threading.Thread):
def run(self):
s.acquire() # 占坑
print("蹲坑 - " + self.name + " - " + str(time.ctime()))
time.sleep(random.randrange(1, 3))
print("走人 - " + self.name + " - " + str(time.ctime()))
s.release() # 走人
if __name__ == '__main__':
for i in range(10):
human = Human()
human.start()
运行结果如下：
蹲坑 - Thread-1 - Tue Jul 17 19:59:15 2018
蹲坑 - Thread-2 - Tue Jul 17 19:59:15 2018
蹲坑 - Thread-3 - Tue Jul 17 19:59:15 2018
蹲坑 - Thread-4 - Tue Jul 17 19:59:15 2018
蹲坑 - Thread-5 - Tue Jul 17 19:59:15 2018
走人 - Thread-1 - Tue Jul 17 19:59:16 2018
蹲坑 - Thread-6 - Tue Jul 17 19:59:16 2018
走人 - Thread-2 - Tue Jul 17 19:59:16 2018
走人 - Thread-3 - Tue Jul 17 19:59:16 2018
蹲坑 - Thread-8 - Tue Jul 17 19:59:16 2018
走人 - Thread-5 - Tue Jul 17 19:59:16 2018
蹲坑 - Thread-7 - Tue Jul 17 19:59:16 2018
蹲坑 - Thread-9 - Tue Jul 17 19:59:16 2018
走人 - Thread-4 - Tue Jul 17 19:59:17 2018
蹲坑 - Thread-10 - Tue Jul 17 19:59:17 2018
走人 - Thread-6 - Tue Jul 17 19:59:17 2018
走人 - Thread-8 - Tue Jul 17 19:59:17 2018
走人 - Thread-9 - Tue Jul 17 19:59:17 2018
走人 - Thread-7 - Tue Jul 17 19:59:18 2018
走人 - Thread-10 - Tue Jul 17 19:59:19 2018
⑦ 通用的条件变量(Event)
Python提供的「用于线程间通信的信号标志」，一个线程标识了一个事件，其他线程处于等待状态，直到事件发生后，所有线程都会被激活。Event对象属性实现了简单的线程通信机制，提供了设置信号，清除信号，等待等用于实现线程间的通信。提供以下四个可供调用的方法：
- is_set()：判断内部标志是否为真
- set()：设置信号标志为真
- clear()：清除Event对象内部的信号标志(设置为false)
- wait(timeout=None)：使线程一直处于堵塞，知道标识符变为True
使用代码示例（汽车等红绿灯的例子）：
import threading
import time
import random
class CarThread(threading.Thread):
def __init__(self, event):
threading.Thread.__init__(self)
self.threadEvent = event
def run(self):
# 休眠模拟汽车先后到达路口时间
time.sleep(random.randrange(1, 10))
print("汽车 - " + self.name + " - 到达路口...")
self.threadEvent.wait()
print("汽车 - " + self.name + " - 通过路口...")
if __name__ == '__main__':
light_event = threading.Event()
# 假设有20台车子
for i in range(20):
car = CarThread(event=light_event)
car.start()
while threading.active_count() > 1:
light_event.clear()
print("红灯等待...")
time.sleep(3)
print("绿灯通行...")
light_event.set()
time.sleep(2)
运行结果如下：
红灯等待...
汽车 - Thread-10 - 到达路口...
汽车 - Thread-14 - 到达路口...
汽车 - Thread-9 - 到达路口...
汽车 - Thread-11 - 到达路口...
汽车 - Thread-12 - 到达路口...
绿灯通行...
汽车 - Thread-11 - 通过路口...
汽车 - Thread-10 - 通过路口...
汽车 - Thread-9 - 通过路口...
汽车 - Thread-14 - 通过路口...
汽车 - Thread-12 - 通过路口...
汽车 - Thread-6 - 到达路口...
汽车 - Thread-6 - 通过路口...
⑧ 定时器(Timer)
和Thread类似，只是要等待一段时间后才会开始运行，单位秒，用法也很简单，
代码示例如下：
import threading
import time
def skill_ready():
print("菜肴制作完成！！！")
if __name__ == '__main__':
t = threading.Timer(5, skill_ready)
t.start()
while threading.active_count() > 1:
print("======菜肴制作中======")
time.sleep(1)
运行结果如下：
======菜肴制作中======
======菜肴制作中======
======菜肴制作中======
======菜肴制作中======
======菜肴制作中======
菜肴制作完成！！！
⑨ 栅栏(Barrier)
Barrier直译栅栏，感觉不怎么好理解，我们可以把它看做是赛马用的栅栏，然后马(线程)依次来到栅栏前等待(wait)，直到所有的马都停在栅栏面前了，然后所有马开始同时出发(start)。简单点说就是： 多个线程间的相互等待，调用了 wait()方法 的线程进入堵塞， 直到所有的线程都调用了wait()方法，然后所有线程同时进入就绪状态， 等待调度运行。
构造函数： Barrier(parties,action=None,timeout=None)
参数解释：
- parties：创建一个可容纳parties条线程的栅栏；
- action：全部线程被释放时可被其中一条线程调用的可调用对象；
- timeout：线程调用wait()方法时没有显式设定timeout，就用的这个作为默认值；
相关属性与函数：
- wait(timeout=None)：表示线程就位，返回值是一个0到parties-1之间的
整数， 每条线程都不一样，这个值可以用作挑选一条线程做些清扫工作，另外如果
你在构造函数里设置了action的话，其中一个线程在释放之前将会调用它。如果调用
出错的话，会让栅栏进入broken状态，超时同样也会进入broken状态，如果栅栏在
处于broke状态的时候调用reset函数，会抛出一个BrokenBarrierError异常。- reset()：本方法将栅栏置为初始状态，即empty状态。所有已经在等待的线程都会
接收到BrokenBarrierError异常，注意当有其他处于unknown状态的线程时，调用
此方法将可能获取到额外的访问。因此如果一个栅栏进入了broken状态， 最好是
放弃他并新建一个栅栏，而不是调用reset方法。- abort()：将栅栏置为broken状态。本方法将使所有正在等待或将要调用
wait()方法的线程收到BrokenBarrierError异常。本方法的使用情景为，比如：
有一条线程需要abort()，又不想给其他线程造成死锁的状态，或许设定
timeout参数要比使用本方法更可靠。- parites：将要使用本 barrier 的线程的数量
- n_waiting：正在等待本 barrier 的线程的数量
- broken：栅栏是否为broken状态，返回一个布尔值
- BrokenBarrierError：RuntimeError的子类，当栅栏被reset()或broken时引发；
使用代码示例如下（公司一起去旅游等人齐才出发）：
import threading
import time
import random
class Staff(threading.Thread):
def __init__(self, barriers):
threading.Thread.__init__(self)
self.barriers = barriers
def run(self):
print("员工 【" + self.name + "】" + "出门")
time.sleep(random.randrange(1, 10))
print("员工 【" + self.name + "】" + "已签到")
self.barriers.wait()
def ready():
print(threading.current_thread().name + "：人齐，出发，出发～～～")
if __name__ == '__main__':
print("要出去旅游啦，大家快集合～")
b = threading.Barrier(10, action=ready, timeout=20)
for i Python小白到老司机，快跟我上车！基础篇(十六)