Python多线程之间同步总结

• Z时代
• 2024-01-10
• 分类：综合

线程安全

多线程主要是为了提高我们cpu的资源使用率。但同时，这会给我们带来很多安全问题！

如果我们在单线程中以“顺序”(串行-->独占)的方式执行代码是没有任何问题的。但是到了多线程的环境下(并行)，如果没有设计和控制得好，就会给我们带来很多意想不到的状况，也就是线程安全性问题。

因为在多线程的环境下，线程是交替执行的，一般他们会使用多个线程执行相同的代码。如果在此相同的代码里边有着共享的变量，或者一些组合操作（访问共享的内存），我们想要的正确结果就很容易出现了问题。

那么到底什么是线程安全呢？

线程安全的问题在于多个线程访问共享的内存而产生的，也就是我们要确保在多线程访问的时候，我们的程序还能按照我们预期的行为去执行，这句话里最重要的是多线程，共享内存，你现在要记住这两句话。

举个例子：

from threading import Thread
num = 0
def add_num():
    global num
    for i in range(100000):
        num += 1
if __name__ == \'__main__\':
    threads = []  # 创建进程列表
    for i in range(0,10):
        t = Thread(target=add_num)
        t.start()
        threads.append(t)
    # 主线程等待所有子线程运行结束
    for thread in threads:
        thread.join()
    print(num)

上面程序是在多线程环境下跑起来，它的num值计算就不对了！我这里的计算结果为793182。

首先，它共享了num这个变量，其次来说num+=1这个操作来说;这是一个组合的操作(注意，它并非是原子性）

实际上的操作是这样子的：

1.读取num 值
2.将值+1
3.将计算结果写入num

于是多线程执行的时候很可能就会有这样的情况：

如果当线程A读取到num 的值是8的时候，同时线程B也进去这个方法上了，也是读取到num 的值为8

它俩都对值进行加1

都将计算结果写入到num上。但是，写入到num上的结果是9

也就是说：两个线程进来了，但是正确的结果是应该返回10，而它返回了9，这是错误的，而我们希望的是num的值为10。

如果说：当多个线程访问某个类的时候，这个类始终能表现出正确的行为，那么这个类就是线程安全的！

临界区

我们把对共享内存进行访问的程序片段称作临界区，当多个线程访问涉及共享内存，共享文件、共享任何资源的情况都会引起发生错，避免错误的关键要找到途径阻止多个进程同时读写共享的数据，在编程中，就是阻止多个线程同时访问临界区。
到了这里，线程安全就是防止多个线程同时访问共享内存，线程安全的本质是其他内存安全。

如何解决线程安全

为了保证使用共享数据的并发编程能正确、高效协作，对于一个好的解决方案，需要满足以下四个条件：

• 任何俩个线程不能同时处于临界区
• 不应对cpu的速度和数量做任何假设
• 临界区外运行的线程不能阻塞其他线程
• 不得使线程无限期等待进入临界区

注意：可以把这里的进程看成线程，进程之间同步的方式和线程类似。

解决方案：

• 互斥量
• 信号量
• 事件

信号量（Python Semaphore对象）

Semaphore对象内部管理一个计数器，该计数器由每个acquire()调用递减，并由每个release()调用递增。计数器永远不会低于零，当acquire()发现计数器为零时，线程阻塞，等待其他线程调用release()。
Semaphore对象支持上下文管理协议。

此计数器是线程共享的，每个线程都可以操作此计数器。

方法：

Semaphore(value=1)：创建一个计数器对象，默认值为1。

acquire(blocking=True, timeout=None)
获取信号，使计数器递减1。
当blocking=True时：如果调用时计数器大于零，则将其减1并立即返回。如果在调用时计数器为零，则阻塞并等待，直到其他线程调用release()使其大于零。这是通过适当的互锁来完成的，因此如果多个acquire()被阻塞，release()将只唤醒其中一个，这个过程会随机选择一个，因此不应该依赖阻塞线程的被唤醒顺序。返回值为True。
当blocking=False时，不会阻塞。如果调用acquire()时计数器为零，则会立即返回False.
如果设置了timeout参数，它将阻塞最多timeout秒。如果在该时间段内没有获取锁，则返回False，否则返回True。

release()
释放信号，使计数器递增1。当计数器为零并有另一个线程等待计数器大于零时，唤醒该线程。

总结：信号量基于计数器来实现线程同步，此计数器为线程之间共享，利用信号量，你可以实现限量线程同时访问临界区，这句话怎么理解，因为你可以设置不同大小的value即可实现。

来看下面的代码：

import time
import threading
def foo():
    time.sleep(2)    #程序休息2秒
    print("ok",time.ctime())
for i in range(20):
    t1=threading.Thread(target=foo,args=())    #实例化一个线程
    t1.start()    #启动线程

执行结果：

ok Tue Jul 18 20:05:58 2017
ok Tue Jul 18 20:05:58 2017
ok Tue Jul 18 20:05:58 2017
ok Tue Jul 18 20:05:58 2017
ok Tue Jul 18 20:05:58 2017
ok Tue Jul 18 20:05:58 2017
ok Tue Jul 18 20:05:58 2017
ok Tue Jul 18 20:05:58 2017
ok Tue Jul 18 20:05:58 2017
ok Tue Jul 18 20:05:58 2017
ok Tue Jul 18 20:05:58 2017
ok Tue Jul 18 20:05:58 2017
ok Tue Jul 18 20:05:58 2017
ok Tue Jul 18 20:05:58 2017
ok Tue Jul 18 20:05:58 2017
ok Tue Jul 18 20:05:58 2017
ok Tue Jul 18 20:05:58 2017
ok Tue Jul 18 20:05:58 2017
ok Tue Jul 18 20:05:58 2017
ok Tue Jul 18 20:05:58 2017

可以看到，程序会在很短的时间内生成20个线程来打印一句话。

这时候就可以为这段程序添加一个信号量的计数器功能，来限制一个时间点内的线程数量。

代码如下：

import time
import threading
s1=threading.Semaphore(5)    #添加一个计数器
def foo():
    s1.acquire()    #计数器获得锁
    time.sleep(2)    #程序休眠2秒
    print("ok",time.ctime())
    s1.release()    #计数器释放锁
for i in range(20):
    t1=threading.Thread(target=foo,args=())    #创建线程
    t1.start()    #启动线程    

执行结果：

ok Tue Jul 18 20:04:38 2017
ok Tue Jul 18 20:04:38 2017
ok Tue Jul 18 20:04:38 2017
ok Tue Jul 18 20:04:38 2017
ok Tue Jul 18 20:04:38 2017
ok Tue Jul 18 20:04:40 2017
ok Tue Jul 18 20:04:40 2017
ok Tue Jul 18 20:04:40 2017
ok Tue Jul 18 20:04:40 2017
ok Tue Jul 18 20:04:40 2017
ok Tue Jul 18 20:04:42 2017
ok Tue Jul 18 20:04:42 2017
ok Tue Jul 18 20:04:42 2017
ok Tue Jul 18 20:04:42 2017
ok Tue Jul 18 20:04:42 2017
ok Tue Jul 18 20:04:44 2017
ok Tue Jul 18 20:04:44 2017
ok Tue Jul 18 20:04:44 2017
ok Tue Jul 18 20:04:44 2017
ok Tue Jul 18 20:04:44 2017

可以看到每隔两秒钟就有五条信息产生，对应了五个线程，我们实现了限量线程访问临界区。

事件（Python Event对象）

事件对象管理一个内部标志，通过set()方法将其设置为True，并使用clear()方法将其设置为False。wait()方法阻塞，直到标志为True。该标志初始为False。

方法：
is_set()
当且仅当内部标志为True时返回True。

set()
将内部标志设置为True。所有等待它成为True的线程都被唤醒。当标志保持在True的状态时，线程调用wait()是不会阻塞的。

clear()
将内部标志重置为False。随后，调用wait()的线程将阻塞，直到另一个线程调用set()将内部标志重新设置为True。

wait(timeout=None)
阻塞直到内部标志为真。如果内部标志在wait()方法调用时为True，则立即返回。否则，则阻塞，直到另一个线程调用set()将标志设置为True，或发生超时。
该方法总是返回True，除非设置了timeout并发生超时。

总结：Event是一个能在多线程中共用的对象，这和信号量是相同的，信号量基于计数器，而event基于共享标志位，一开始它包含一个为 False的信号标志，一旦在任一一个线程里面把这个标记改为 True，那么所有的线程都会看到这个标记变成了 True。看到这里，event要么把共享此标志的线程全部给阻塞，要么大家一起运行。

设想这样一个场景：

你创建了10个子线程，每个子线程分别爬一个网站，一开始所有子线程都是阻塞等待。一旦某个事件发生：例如有人在网页上点了一个按钮，或者某人在命令行输入了一个命令，10个爬虫同时开始工作。

代码片段可以简写为：

import threading
import time
class spider(threading.Thread):
 def __init__(self, n, event):
        super().__init__()
        self.n = n
        self.event = event
 def run(self):
     print(f\'第{self.n}号爬虫已就位！\')
     self.event.wait()
     print(f\'信号标记变为True！！第{self.n}号爬虫开始运行\')
eve = threading.Event()
for num in range(10):
    crawler = spider(num, eve)
    crawler.start()
input(\'按下回车键，启动所有爬虫！\')
eve.set() #设置标志为true
time.sleep(5)

运行结果如图所示：

互斥量

• 原始锁 （Python Lock对象）
• 重入锁 （Python RLock对象）
• 条件锁 （Python Condition(条件对象)
• 障碍锁 (Python Barrier对象)

原始锁 （Python Lock对象）

方法：

acquire(blocking=True,timeout=-1)：默认阻塞，阻塞可以设置超时时间。非阻塞时，timeout禁止设置。成功获取锁，返回True，否则返回False

release( )：释放锁。可以从任何线程释放。已上锁的锁，会抛出RuntimeError异常。

总结：原始锁是一个处于两个状态的变量：解锁和加锁，只需要一个二进制位表示，原始锁和前面信号量和event都是可以被线程之间共享的，这里原始锁对象始终只让一个线程进入临界区，如果该锁在进入临界区之前加锁，则其他线程在进入此临界区时都会被阻塞，等待临界区的线程完成解锁操作。其他线程才有机会进入此临界区（随机选择一个线程允许它获得锁）。

原始锁适用于访问和修改同一个资源的时候，引起资源争用的情况下。使用锁的注意事项：锁的使用场景：

1.少用锁，除非有必要。多线程访问加锁的资源时，由于锁的存在，实际就变成了串行。

2.加锁时间越短越好，不需要就立即释放锁。

3.一定要避免死锁，使用with或者try...finally。

我们修改了最开始的例子，使用原始锁保证了线程安全：

import threading
num = 0
lock = threading.Lock()
def add_num():
    global num
    try:
        lock.acquire()
        for i in range(100000):
            num += 1
    finally:
        lock.release()
if __name__ == \'__main__\':
    threads = []  # 创建进程列表
    for i in range(0,10):
        t = threading.Thread(target=add_num)
        t.start()
        threads.append(t)
    # 主线程等待所有子线程运行结束
    for thread in threads:
        thread.join()
    print(num)

重入锁  （Python RLock对象）

　　为了支持在同一线程中多次请求同一资源，python提供了“可重入锁”：threading.RLock。RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源可以被多次require。直到一个线程所有的acquire都被release，其他的线程才能获得资源。

threading.Rlock() 允许多次锁资源，acquire() 和 release() 必须成对出现，也就是说加了几把锁就得释放几把锁。

方法：

acquire(blocking=True,timeout = -1)：内部counter变量加1

release():内部counter变量减1

lock = threading.Lock()
# 死锁
lock.acquire()
lock.acquire()
print(\'...\')
lock.release()
lock.release()
/////////////////////////
rlock = threading.RLock()
# 同一线程内不会阻塞线程，可以多次acquire
rlock.acquire()
rlock.acquire()
print(\'...\')
rlock.release()
rlock.release()

条件锁  （Python Condition(条件对象)

条件变量允许一个或多个线程等待，直到他们被另一个线程通知。

方法：

acquire(*args)
获取锁。这个方法调用底层锁的相应方法。

release()
释放锁。这个方法调用底层锁的相应方法。

wait(timeout=None)
线程挂起，等待被唤醒(其他线程的notify方法)或者发生超时。调用该方法的线程必须先获得锁，否则引发RuntimeError。
该方法会释放底层锁，然后阻塞，直到它被另一个线程中的相同条件变量的notify()或notify_all()方法唤醒，或者发生超时。一旦被唤醒或超时，它会重新获取锁并返回。
返回值为True，如果给定timeout并发生超时，则返回False。

wait_for(predicate, timeout=None)
等待知道条件变量的返回值为True。predicate应该是一个返回值可以解释为布尔值的可调用对象。可以设置timeout以给定最大等待时间。
该方法可以重复调用wait()，直到predicate的返回值解释为True，或发生超时。该方法的返回值就是predicate的最后一个返回值，如果发生超时，返回值为False。
如果忽略超时功能，该方法大致相当于：

while not predicate():
	con.wait()

它与wait()的规则相同：调用前必须先获取锁，阻塞时释放锁，并在被唤醒时重新获取锁并返回。

notify(n=1)
默认情况下，唤醒等待此条件变量的一个线程(如果有)。调用该方法的线程必须先获得锁，否则引发RuntimeError。
该方法最多唤醒n个等待中的线程，如果没有线程在等待，它就是要给无动作的操作。
注意：要被唤醒的线程实际上不会马上从wait()方法返回(唤醒)，而是等到它重新获取锁。这是因为notify()并不会释放锁，需要线程本身来释放(通过wait()或者release())

notify_all()
此方法类似于notify()，但唤醒的时所有等待的线程。

障碍锁 (Python Barrier对象)

Barrier(parties, action=None, timeout=None)

每个线程通过调用wait()尝试通过障碍，并阻塞，直到阻塞的数量达到parties时，阻塞的线程被同时全部释放。

action是一个可调用对象，当线程被释放时，其中一个线程会首先调用action，之后再跑自己的代码。

timeout时默认的超时时间。

方法：

wait(timeout=None)

尝试通过障碍并阻塞。

返回值是一个在0到parties-1范围内的整数，每个线程都不同。

其中一个线程在释放之前将调用action。如果此调用引发错误，则障碍将进入断开状态。

如果等待超时，障碍也将进入断开状态。

如果在线程等待期间障碍断开或重置，此方法可能会引发BrokenBarrierError错误。

reset()

重置障碍，返回默认的空状态，即当前阻塞的线程重新来过。见例二

abort()

将障碍置为断开状态，这将导致已调用wait()或之后调用wait()引发BrokenBarrierError。见例三

属性：

partier

通过障碍所需的线程数。

n_waiting

当前在屏障中等待的线程数

broken

如果屏障处于断开状态，则返回True。

总结:障碍锁也是线程之间共享的，可以阻塞一个或者多个线程，不同的是，当阻塞的线程达到一个目标值以后，被阻塞的线程会被同时释放，提供了action，在达到目标值以后会执行此action可调用对象（用户自定义函数），且只执行一次。

实例

例一：

# -*- coding:utf-8 -*-
import threading
import time
def open():
    print(\'人数够了， 开门!\')
barrier = threading.Barrier(3, open)
class Customer(threading.Thread):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.n = 3
    def run(self):
        while self.n > 0:
            self.n -= 1
            print(\'{0}在等着开门.\'.format(self.name))
            try:
                barrier.wait(2)
            except threading.BrokenBarrierError:
                pass
            print(\'开门了， go go go\')
if __name__ == \'__main__\':
    t1 = Customer(name=\'A\')
    t2 = Customer(name=\'B\')
    t3 = Customer(name=\'C\')
    t1.start()
    t2.start()
    t3.start()

运行结果：

A在等着开门.
B在等着开门.
C在等着开门.
人数够了， 开门!
开门了， go go go
开门了， go go go
开门了， go go go
C在等着开门.
A在等着开门.
B在等着开门.
人数够了， 开门!
开门了， go go go
开门了， go go go
开门了， go go go
...

例二：

# -*- coding:utf-8 -*-
import threading
import time
def open():
    print(\'人数够了， 开门!\')
barrier = threading.Barrier(3, open)
class Customer(threading.Thread):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.n = 3
    def run(self):
        while self.n > 0:
            self.n -= 1
            print(\'{0}在等着开门.\'.format(self.name))
            try:
                barrier.wait(2)
            except threading.BrokenBarrierError:
                continue
            print(\'开门了， go go go\')
class Manager(threading.Thread):
    def run(self):
        print(\'前面几个排队的不算，重新来\')
        barrier.reset()
if __name__ == \'__main__\':
    t1 = Customer(name=\'A\')
    t2 = Customer(name=\'B\')
    t3 = Customer(name=\'C\')
    tm = Manager()
    t1.start()
    t2.start()
    tm.start()
    t3.start()

运行结果：

A在等着开门.
B在等着开门.
前面几个排队的不算，重新来
A在等着开门.
B在等着开门.
C在等着开门.
人数够了， 开门!
开门了， go go go
开门了， go go go
开门了， go go go
A在等着开门.
C在等着开门.
B在等着开门.
人数够了， 开门!
开门了， go go go
开门了， go go go
开门了， go go go
C在等着开门.

例三：

# -*- coding:utf-8 -*-
import threading
def open():
    print(\'人数够了， 开门!\')
barrier = threading.Barrier(3, open)
class Customer(threading.Thread):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.n = 3
    def run(self):
        while self.n > 0:
            self.n -= 1
            print(\'{0}在等着开门.\'.format(self.name))
            try:
                barrier.wait(2)
            except threading.BrokenBarrierError:
                print(\'今天好像不开门了，回家.\')
                break
            print(\'开门了， go go go\')
class Manager(threading.Thread):
    def run(self):
        print(\'老板跟小姨子跑了，不开门了！\')
        barrier.reset()
if __name__ == \'__main__\':
    t1 = Customer(name=\'A\')
    t2 = Customer(name=\'B\')
    t3 = Customer(name=\'C\')
    tm = Manager()
    t1.start()
    t2.start()
    tm.start()
    t3.start()

运行结果：

A在等着开门.
B在等着开门.
老板跟小姨子跑了，不开门了！
今天好像不开门了，回家.
今天好像不开门了，回家.
C在等着开门.
今天好像不开门了，回家.

参考：

python中的线程之semaphore信号量

一日一技：Python多线程的事件监控

Python多线程-Barrier(障碍对象)

python高级-多线程总结（思维导图）

以上是 Python多线程之间同步总结 的全部内容， 来源链接： utcz.com/z/386478.html