python 线程锁

目录

1、线程安全介绍

2、threading5种常见锁

2.1 同步锁（互斥锁）Lock

(1) 基本介绍及相关方法

(2) 给案例加lock锁

（3）with语句

2.2 递归锁Rlock

（1）基本介绍及相关方法

(2) 给案例加递归锁

(3) with 语句

2.3 条件锁Condition

(1)基本介绍和方法

（2）with语句

（3）案例

2.4 事件锁Event

（1）基本介绍和方法

（2）使用方式

（3）案例

2.5 信号量锁Semaphore

（1）基本介绍和方法

（2）案例

（3）with语句

3、死锁现象

4、总结

1、线程安全介绍

线程安全的问题最主要还是由线程切换导致的。

线程安全是多线程或多进程编程中的一个概念，在拥有共享数据的多条线程并行执行的程序中，线程安全的代码会通过同步机制保证各个线程都可以正常且正确的执行，不会出现数据污染等意外情况。

线程不安全案例：

创建2个线程：线程1对num进行一千万次+1的操作；线程2对num进行一千万次-1的操作

import threading

num = 0

def add():
    global num
    for i in range(10_000_000):
        num += 1

def sub():
    global num
    for i in range(10_000_000):
        num -= 1

if __name__ == "__main__":
    subThread01 = threading.Thread(target=add)
    subThread02 = threading.Thread(target=sub)

    subThread01.start()
    subThread02.start()

    subThread01.join()
    subThread02.join()

    print("num result : %s" % num)

运行三次结果分别如下：

num result : 645898
num result : -331131
num result : 901713

num最后并不是我们所想象的结果0。因此需要通过锁来保障线程切换的时机。

2、threading5种常见锁

threading模块中提供了5种最常见的锁：

• 同步锁：lock（一次只能放行一个）

• 递归锁：rlock（一次只能放行一个）

• 条件锁：condition（一次可以放行任意个）

• 事件锁：event（一次全部放行）

• 信号量锁：semaphore（一次可以放行特定个）

2.1 同步锁（互斥锁）Lock

(1) 基本介绍及相关方法

• 互斥指的是某一资源同一时刻仅能有一个访问者对其进行访问，具有唯一性和排他性，但是互斥无法限制访问者对资源的访问顺序，即访问是无序的
• 同步是指在互斥的基础上（大多数情况），通过其他机制实现访问者对资源的有序访问
• 同步其实已经实现了互斥，是互斥的一种更为复杂的实现，因为它在互斥的基础上实现了有序访问的特点

# 生成互斥锁对象
lock = threading.Lock()

# 获取锁，未获取到程序将会阻塞(当一个线程在执行被上锁的代码块时，将不允许切换到其他线程)
lock.acquire()

# 释放锁
lock.release()

# 判断该锁对象是否处于上锁状态
lock.locked()

(2) 给案例加lock锁

import threading

num = 0
lock = threading.Lock()  # 生成互斥锁对象

def add():
    global num
    lock.acquire()   # 上锁
    for i in range(10_000_000):
        num += 1
    lock.release()   # 释放锁

def sub():
    global num
    lock.acquire()  # 上锁
    for i in range(10_000_000):
        num -= 1
    lock.release()    # 释放锁

if __name__ == "__main__":
    subThread01 = threading.Thread(target=add)
    subThread02 = threading.Thread(target=sub)

    subThread01.start()
    subThread02.start()

    subThread01.join()
    subThread02.join()

    print("num result : %s" % num)

执行上述代码，每次结果均为0。

上述代码完全变成了串行的状态，对于这种计算密集型I/O业务来说，还不如直接使用串行化单线程执行来得快，所以这个例子仅作为一个示例，不能概述锁真正的用途。

（3）with语句

由于threading.Lock()对象中实现了enter__()与__exit()方法，故可以使用with语句进行上下文管理形式的加锁解锁操作：

import threading

num = 0
lock = threading.Lock()


def add():
    with lock:
        # 自动加锁
        global num
        for i in range(10_000_000):
            num += 1
        # 自动解锁


def sub():
    with lock:
        # 自动加锁
        global num
        for i in range(10_000_000):
            num -= 1
        # 自动解锁


if __name__ == "__main__":
    subThread01 = threading.Thread(target=add)
    subThread02 = threading.Thread(target=sub)

    subThread01.start()
    subThread02.start()

    subThread01.join()
    subThread02.join()

    print("num result : %s" % num)

上述代码与(2)中代码效果一样。

2.2 递归锁Rlock

（1）基本介绍及相关方法

        有时候在一些逻辑当中，可能需要多次获取锁也就是嵌套锁，如果这种情况还是使用互斥锁的话就会造成死锁，这个时候就应该用到可重入锁(递归锁) RLock()。

        递归锁是同步锁的一个升级版本，在同步锁的基础上可以做到连续重复使用多次acquire()后再重复使用多次release()的操作，但是一定要注意加锁次数和解锁次数必须一致，否则也将引发死锁现象。

lock = threading.RLock()   # 返回一个递归锁对象

# 获取锁，未获取到程序将会阻塞(当一个线程在执行被上锁的代码块时，将不允许切换到其他线程)
lock.acquire()

# 释放锁
lock.release()

# 判断该锁对象是否处于上锁状态，返回布尔值
lock.locked()

(2) 给案例加递归锁

        下面这段操作如果使用同步锁则会发生死锁现象，因为第二次去获得锁的时候锁已经被拿走了，锁还没有释放又再一次请求，这样就形成死锁了。

        但是递归锁不会，递归锁在同一个线程里面可以连续多次调用 acquire，除了这一点，可重入锁和互斥锁没有其他区别。

import threading

num = 0
lock = threading.RLock()   # 返回一个递归锁对象


def add():
    lock.acquire()
    lock.acquire()
    global num
    for i in range(10_000_000):
        num += 1
    lock.release()
    lock.release()


def sub():
    lock.acquire()
    lock.acquire()
    global num
    for i in range(10_000_000):
        num -= 1
    lock.release()
    lock.release()


if __name__ == "__main__":
    subThread01 = threading.Thread(target=add)
    subThread02 = threading.Thread(target=sub)

    subThread01.start()
    subThread02.start()

    subThread01.join()
    subThread02.join()

    print("num result : %s" % num)

(3) with 语句

由于threading.RLock()对象中实现了enter__()与__exit()方法，故可以使用with语句进行上下文管理形式的加锁解锁操作：

import threading

num = 0
lock = threading.RLock()   # 返回一个递归锁对象

def add():
    with lock:
        global num
        for i in range(10_000_000):
            num += 1

def sub():
    with lock:
        global num
        for i in range(10_000_000):
            num -= 1

if __name__ == "__main__":
    subThread01 = threading.Thread(target=add)
    subThread02 = threading.Thread(target=sub)

    subThread01.start()
    subThread02.start()

    subThread01.join()
    subThread02.join()

    print("num result : %s" % num)

2.3 条件锁Condition

(1)基本介绍和方法

Condition 是 python 中的条件锁，内部也是通过 Lock() 和 RLock() 锁实现，所以具有他们的阻塞特性， 并且在这基础上增加了暂停线程运行的功能，所以可以用来同步复杂的线程间通信。

# 生成一个条件锁对象
cond = threading.Condition()

# 上锁
cond.acquire()

# 解锁
cond.release()

# 挂起线程，直到收到一个 notify 通知才会被唤醒
cond.wait()

# 唤醒一个 Condition 的 waiting 池中的线程
cond.notify()

# 唤醒所有 Condition 的 waiting 池中的线程
cond.notify_all()

（2）with语句

由于threading.Condition()对象中实现了enter__()与__exit()方法，故可以使用with语句进行上下文管理形式的加锁解锁操作。

（3）案例

该案例使用with语句实现加锁解锁。

import threading

class Zhou(threading.Thread):
    def __init__(self, cond):
        super().__init__(name="周杰伦")
        self.cond = cond
                 
    def run(self):
        with self.cond:
            # Condition()对象中也实现了__enter__()与__exit__()魔法方法，所以也是可以通过 with 语句调用的
            print("{}: 海平面远方开始阴霾, 悲伤要怎么平静纯白".format(self.name))
            self.cond.notify()
            self.cond.wait()

            print("{}: 你用唇语说你要离开, 那难过无声慢了下来 ".format(self.name))
            self.cond.notify()
            self.cond.wait()

            print("{}: 转身离开, 你有话说不出来 ".format(self.name))
            self.cond.notify()
            self.cond.wait()

            print("{}: 我们的爱, 差异一直存在, 等待竟累积成伤害 ".format(self.name))
            self.cond.notify()
            self.cond.wait()

            print("{}: 蔚蓝的珊瑚海, 错过瞬间苍白 ".format(self.name))
            self.cond.notify()
            self.cond.wait()

            print("{}: 热情不再, 笑容勉强不来, 爱深埋珊瑚海".format(self.name))
        
                         


class Liang(threading.Thread):
    def __init__(self, cond):
        super().__init__(name="梁心颐")
        self.cond = cond
        
    def run(self):
        with self.cond:
            # 在调用with cond 之后才能调用 wait 或者 notify 方法
            self.cond.wait()
            print("{}: 我的脸上始终挟带, 一抹浅浅的无奈".format(self.name))
            self.cond.notify()

            self.cond.wait()
            print("{}: 汹涌潮水, 你听明白, 不是浪而是泪海 ".format(self.name))
            self.cond.notify()

            self.cond.wait()
            print("{}: 海鸟跟鱼相爱, 只是一场意外 ".format(self.name))
            self.cond.notify()

            self.cond.wait()
            print("{}: 转身离开, 分手说不出来 ".format(self.name))
            self.cond.notify()

            self.cond.wait()
            print("{}: 当初彼此, 不够成熟坦白".format(self.name))
            self.cond.notify()
                         


if __name__ == "__main__":
    cond = threading.Condition()
    liang = Liang(cond)
    zhou = Zhou(cond)
    # 这里的启动顺序很重要
    liang.start()
    zhou.start()

两个线程通过 cond 通信，阻塞自己并唤醒对方，从而达到有序执行，打印结果如下：

周杰伦: 海平面远方开始阴霾, 悲伤要怎么平静纯白
梁心颐: 我的脸上始终挟带, 一抹浅浅的无奈
周杰伦: 你用唇语说你要离开, 那难过无声慢了下来 
梁心颐: 汹涌潮水, 你听明白, 不是浪而是泪海 
周杰伦: 转身离开, 你有话说不出来 
梁心颐: 海鸟跟鱼相爱, 只是一场意外 
周杰伦: 我们的爱, 差异一直存在, 等待竟累积成伤害 
梁心颐: 转身离开, 分手说不出来 
周杰伦: 蔚蓝的珊瑚海, 错过瞬间苍白 
梁心颐: 当初彼此, 不够成熟坦白
周杰伦: 热情不再, 笑容勉强不来, 爱深埋珊瑚海

2.4 事件锁Event

（1）基本介绍和方法

Event 是 python 中的事件锁，它是根据条件锁来实现的，条件锁一次可以放行任意个子线程，而事件锁一次只能放行全部。

Event 对象中有一个信号标志，默认为 False，可以把这个标志看做为一个红绿灯，当红灯时所有子线程都暂停运行，并进入“等待”状态，当绿灯时所有子线程都恢复“运行”。

如果一个线程等待一个 Event 对象，那么这个 Event 对象的标志将决定这个线程是否会被阻塞。需要注意的是，如果一个线程将 Event 对象的标志设置为真，那么所有等待这个 Event 对象的线程都将会被放行。

# 生成一个事件锁对象
eve = threading.Event()


# 将事件锁设置为红灯状态,即所有线程暂停运行
eve.clear()
 
# 判断事件锁的状态，红灯为False,绿灯为True
eve.is_set()


# 将当前线程设置’等待‘状态，只有该线程接到"绿灯通知"或者超时时间到期之后才会继续运行，
#  在等待状态下的线程将允许系统根据策略自行切换到其他线程中运行
eve.wait(timeout=None) 


# 将事件锁设置为绿灯状态,即所有线程恢复运行
eve.set()

（2）使用方式

事件锁 Event 并没有实现魔法方法 enter() 和 exit()，所以不能通过 with 语句调用，只能按照常规方式。

（3）案例

我们来看一个模拟交通红绿灯的例子

import time
import threading


def light(eve):
    print(f'当前时间:{time.ctime()}, 红灯还有 5s 结束!')
    time.sleep(5)
    print(f'当前时间:{time.ctime()}, 绿灯亮!')
    eve.set()  # 设置事件锁标志为 True


def car(eve, name):
    print(f'当前时间:{time.ctime()}, 车 {name} 正在等红灯')
    eve.wait()  # 将当前线程设置为等待状态，等待事件锁标志为 True 再执行
    print(f'当前时间:{time.ctime()}, 车 {name} 开始通行')


if __name__ == '__main__':
    eve = threading.Event()  # 事件锁默认标志为 False
    t1 = threading.Thread(target=light, args=(eve,))
    t1.start()

    for each in 'ABCDE':
        t2 = threading.Thread(target=car, args=(eve, each))
        t2.start()

执行结果如下：

当前时间:Wed Jun 28 15:15:35 2023, 红灯还有 5s 结束!
当前时间:Wed Jun 28 15:15:35 2023, 车 A 正在等红灯
当前时间:Wed Jun 28 15:15:35 2023, 车 B 正在等红灯
当前时间:Wed Jun 28 15:15:35 2023, 车 C 正在等红灯
当前时间:Wed Jun 28 15:15:35 2023, 车 D 正在等红灯
当前时间:Wed Jun 28 15:15:35 2023, 车 E 正在等红灯
当前时间:Wed Jun 28 15:15:40 2023, 绿灯亮!
当前时间:Wed Jun 28 15:15:40 2023, 车 A 开始通行
当前时间:Wed Jun 28 15:15:40 2023, 车 B 开始通行
当前时间:Wed Jun 28 15:15:40 2023, 车 E 开始通行
当前时间:Wed Jun 28 15:15:40 2023, 车 D 开始通行
当前时间:Wed Jun 28 15:15:40 2023, 车 C 开始通行

可见所有 car 线程启动后，并没有完全执行，而是在 eve.wait() 这里阻塞住了，这个时候事件锁的标志为 False，当5秒后，事件锁执行 eve.set() 将标志设置为 Ture 之后，所有的 car 线程才开始又继续执行。

2.5 信号量锁Semaphore

（1）基本介绍和方法

信号量锁也是根据条件锁来做的，它内部维护了一个计数器，启动线程计数器减1，计数器为 0 的时候线程就会被阻塞，然后线程结束计数器再加1。

它与条件锁和事件锁的区别如下：

• 条件锁：一次可以放行任意个处于“等待”状态的线程

• 事件锁：一次可以放行全部的处于“等待”状态的线程

• 信号量锁：通过规定，成批的放行特定个处于“上锁”状态的线程

# 生成一个信号量锁对象
sem = threading.Semaphore()


# 上锁，内置计数器减一
sem.acquire()


# 解锁，内置计数器加一
sem.release()

（2）案例

假设有11个人需要上厕所，但是厕所里面只有三个坑位，所以一次只能有三个线程并发

import threading
import time


def toilet(sem, name):
    sem.acquire()
    time.sleep(3)
    print(f"当前时间: {time.ctime()}, person_{name} 上厕所完毕!", )
    sem.release()


if __name__ == "__main__":
    sem = threading.Semaphore(3)   # 最多同时运行3个线程
    for i in range(11):
        t = threading.Thread(target=toilet, args=(sem, str(i)))
        t.start()

运行结果如下：

当前时间: Wed Jun 28 15:23:36 2023, person_0 上厕所完毕!
当前时间: Wed Jun 28 15:23:36 2023, person_1 上厕所完毕!
当前时间: Wed Jun 28 15:23:36 2023, person_2 上厕所完毕!
当前时间: Wed Jun 28 15:23:39 2023, person_4 上厕所完毕!
当前时间: Wed Jun 28 15:23:39 2023, person_5 上厕所完毕!
当前时间: Wed Jun 28 15:23:39 2023, person_3 上厕所完毕!
当前时间: Wed Jun 28 15:23:42 2023, person_6 上厕所完毕!
当前时间: Wed Jun 28 15:23:42 2023, person_7 上厕所完毕!
当前时间: Wed Jun 28 15:23:42 2023, person_8 上厕所完毕!
当前时间: Wed Jun 28 15:23:45 2023, person_9 上厕所完毕!
当前时间: Wed Jun 28 15:23:45 2023, person_10 上厕所完毕!

从输出控制台可以看到，同一时间只有3个线程运行，隔3秒之后，再次启动三个线程，直到11个人全部上完厕所为止。

（3）with语句

Semaphore 里面也实现了 __enter__() 与 __exit__() 两个魔法方法，所以也是可以通过 with 语句进行加锁解锁操作，下面的代码与上面的实现效果一样

def toilet(sem, name):
    with sem:
        time.sleep(3)
        print(f"当前时间: {time.ctime()}, person_{name} 上厕所完毕!", )

3、死锁现象

一般来说，形成死锁有两种形式

① 一个线程里面嵌套获取一把互斥锁。

② 多个线程不按顺序同时获取多个锁，例如：线程1嵌套获取ab两个锁，线程2嵌套获取ba两个锁，当线程1获取到了a在等待b，而线程2获取到了b在等待a，两个线程就形成了一个循环等待的圆，从而形成死锁。

对于第一种情况，可以使用可重入锁。

对于第二情况，其实和 mysql 中的死锁性质差不多，因争夺资源而造成的一种死循环，保证多个线程获取锁的顺序是同样的即可避免这种情况。

对于同步锁来说，一次acquire()必须对应一次release()，不能出现连续重复使用多次acquire()后再重复使用多次release()的操作，这样会引起死锁造成程序的阻塞。

4、总结

threading 模块中的5种锁

① 互斥锁：Lock，一次只能放行一个，可以通过 with 语句调用。

② 可重入锁：RLock，一次只能放行一个，可以通过 with 语句调用。

③ 条件锁：Condition，一次可以放行任意个，可以通过 with 语句调用。

④ 事件锁：Event，一次全部放行，不能通过 with 语句调用。

⑤ 信号量锁:semaphore，一次可以放行特定个，可以通过 with 语句调用。