Python多线程同步问题

水深无声 2022-11-09 11:20 198阅读 0赞

## 引言 ##

> 介绍多线程共享全局变量，并研究Python多线程资源竞争导致线程不同步的问题。
> 
> 利用 **线程锁（Lock）** 机制实现线程同步。

## 多线程-共享全局变量 ##

import time
    from threading import Thread
    
    
    g_num = 100
    
    
    def work1():
        global g_num
        for i in range(3):
            g_num += 1
    
        print("----in work1, g_num is %d---" % g_num)
    
    
    def work2():
        global g_num
        print("----in work2, g_num is %d---" % g_num)
    
    
    def main():
        
        print("---线程创建之前g_num is %d---" % g_num)
    
        t1 = Thread(target=work1)
        t1.start()
    
        #延时一会，保证t1线程中的事情做完
        time.sleep(1)
    
        t2 = Thread(target=work2)
        t2.start()
        
     
    if __name__ == "__main__":
        main()

运行结果:

---线程创建之前g_num is 100---
    ----in work1, g_num is 103---
    ----in work2, g_num is 103---

## 列表当做实参传递到线程中 ##

import time
    from threading import Thread
    
    
    def work1(nums):
        nums.append(44)
        print("----in work1---",nums)
    
    
    def work2(nums):
        #延时一会，保证t1线程中的事情做完
        time.sleep(1)
        print("----in work2---",nums)
    
    g_nums = [11,22,33]
    
    t1 = Thread(target=work1, args=(g_nums,))
    t1.start()
    
    t2 = Thread(target=work2, args=(g_nums,))
    t2.start()

运行结果:

----in work1--- [11, 22, 33, 44]
    ----in work2--- [11, 22, 33, 44]

*  在一个进程内的所有线程共享全局变量，很方便在多个线程间共享数据
 *  缺点就是，线程是对全局变量随意遂改可能造成多线程之间对全局变量的混乱（即线程非安全）

## Python 多线程资源竞争问题 ##

我们就用自定义一个自增线程类继承 `threading.Thread` 类来模拟资源竞争问题。

### 代码演示 ###

""" Python 多线程同步问题 """
    import time
    import threading
    
    
    # 线程共享变量
    num1 = 0
    num2 = 0
    
    
    class NumIncrement(threading.Thread):
    	"""自定义自增线程类"""
    
    	def __init__(self, count):
    		super().__init__()
    		self.count = count
    	
    	def run(self):
    		self.num_increment()
    
    	def num_increment(self):
    		"""数字自增"""
    		global num1
    		for i in range(self.count):
    			num1 += 1
    
    
    def sync_test():
    	"""多线程同步测试"""
    	global num1, num2
    
    	print('num1=%d' % num1)
    	print('num2=%d' % num2)
    
    	count = 1000
    	t1 = NumIncrement(count)
    	t2 = NumIncrement(count)
    
    	t1.start()
    	t2.start()
    	t1.join()
    	t2.join()
    
    	# 单线程自增
    	for i in range(2):
    		for j in range(count):
    			num2 += 1
    
    	print('num1=%d' % num1)
    	print('num2=%d' % num2)
    
    
    def main():
    	sync_test()
    
    
    if __name__ == '__main__':
    	main()

### 运行结果 ###

`count = 1000` 运行结果如下：

num1=0
    num2=0
    num1=2000
    num2=2000
    [Finished in 0.1s]

但你把 `count` 设置成 `1000000（1百万）`或者更大试下，你会发现多线程自增得不到正确的数，且每次结果都可能不一样。

`count = 1000000` 运行结果如下：

num1=0
    num2=0
    num1=1377494
    num2=2000000
    [Finished in 0.4s]

**如果多个线程同时对同一个全局变量操作，会出现资源竞争问题，从而数据结果会不正确**

num += 1
    
    会转换成
    
    num = num + 1

### 问题分析 ###

假设当时 `num = 100`，第一个线程抢到时间片运行，执行完 `num + 1` 加法操作，然后刚想执行 `num = 101` 赋值操作时被第二个线程抢到线程执行权，且完整执行完 `num += 1`，由于上个线程还没有完成赋值操作，所以此时 `num = 101`，然后 **切换到第一个线程的上下文环境** 进行赋值操作 `num = 101`，因为第一个线程的加法操作已经完成只要继续进行赋值就行，但此时 `num` 已经被第二个线程赋值变成了 `num = 101` ，所以再让 `num = 101` 已经让 `num` 重复赋值了，因此数据结果会不正确。

为什么 `count = 1000` 比较小的数的时候，很难出现数据结果不正确呢，可能 `cpu` 对这些小数据计算很快，一下子就完成简单的 `+1` 操作。而到了几百万、千万，抢到的时间片可能不够用，无法一次完成全部操作，且非原子性操作，可进行线程的上下文切换。

***原子操作***(`atomic operation`)是不需要 `synchronized` ，这是多线程编程的老生常谈了。所谓 \*\*\*原子操作\*\*\*是指不会被线程调度机制打断的操作；这种操作一旦开始，就一直运行到结束，中间不会有任何 `context switch` （切换到另一个线程的上下文）

`num += 1` 是非原子性操作，要先进行 `num + 1` 加法操作，然后进行 `num` 赋值操作。

## 线程同步 ##

### 同步的概念 ###

同步就是协同步调，按预定的先后次序进行运行。如: **你说完，我再说**。

"同"字从字面上容易理解为一起动作

其实不是，"同"字应是指协同、协助、互相配合。

如进程、线程同步，可理解为进程或 线程 A 和 B 一块配合，A 执行到一定程度时要依靠 B 的某个结果，于是停下来，示意 B 运行，B 执行，再将结果给 A，A 再继续操作。

### 线程锁机制 ###

#### 互斥锁 ####

当多个线程几乎同时修改某一个共享数据的时候，需要进行同步控制

线程同步能够保证多个线程安全访问竞争资源，最简单的同步机制是引入互斥锁。

互斥锁为资源引入一个状态：**锁定 / 非锁定**

某个线程要更改共享数据时，先将其锁定，此时资源的状态为 **锁定**，其他线程不能更改；直到该线程释放资源，将资源的状态变成 **非锁定**，其他的线程才能再次锁定该资源。互斥锁保证了每次只有一个线程进行写入操作，从而保证了多线程情况下数据的准确性。

对于上文提出的那个计算错误的问题，可以通过线程同步来进行解决

思路，如下:

1.  系统调用 t1，然后获取到 g\_num 的值为0，此时上一把锁，即不允许其他线程操作 g\_num
2.  t1 对 g\_num 的值进行+1
3.  t1 解锁，此时 g\_num 的值为1，其他的线程就可以使用 g\_num了，而且是 g\_num 的值不是 0 而是 1
4.  同理其他线程在对 g\_num 进行修改时，都要先上锁，处理完后再解锁，在上锁的整个过程中不允许其他线程访问，就保证了数据的准确性

`threading` 模块中定义了 `Lock` 类，可以方便的处理锁定：

import threading
    
    # 创建锁
    mutex = threading.Lock()
    
    # 锁定
    mutex.acquire()
    
    # 释放
    mutex.release()

""" Python 互斥锁解决多线程资源竞争问题 """
    import time
    import threading
    
    
    # 线程共享变量
    g_num = 0
    
    # 创建一个互斥锁
    # 默认是未上锁的状态
    mutex = threading.Lock()
    
    
    def work1(num):
        global g_num
        for i in range(num):
            mutex.acquire()  # 上锁
            g_num += 1
            mutex.release()  # 解锁
    
        print("---work1---g_num=%d" % g_num)
    
    
    def work2(num):
        global g_num
        for i in range(num):
            mutex.acquire()  # 上锁
            g_num += 1
            mutex.release()  # 解锁
    
        print("---work2---g_num=%d" % g_num)
    
    
    def mutex_test():
        """互斥锁测试"""
    
    	# 创建2个线程，让他们各自对g_num加1000000次
    	count = 1000000
    	t1 = threading.Thread(target=work1, args=(count,))
    	t1.start()
    
    	t2 = threading.Thread(target=work2, args=(count,))
    	t2.start()
    
    	# 等待计算完成
        # len(threading.enumerate()) = 当前程序线程的数量
        # 为1说明只剩下主线程
    	while len(threading.enumerate()) != 1:
    	    time.sleep(1)
    
    	print("2个线程对同一个全局变量操作之后的最终结果是:%s" % g_num)
    
    
    def main():
    	mutex_test()
    
    
    if __name__ == '__main__':
    	main()

运行结果如下：

---work1---g_num=1974653
    ---work2---g_num=2000000
    2个线程对同一个全局变量操作之后的最终结果是:2000000

可以看到最后的结果，加入互斥锁后，其结果与预期相符。

**注意：**

*  如果这个锁之前是没有上锁的，那么 `acquire` 不会堵塞
 *  如果在调用 `acquire` 对这个锁上锁之前 它已经被 其他线程上了锁，那么此时 `acquire` 会堵塞，直到这个锁被解锁为止

#### 死锁 ####

在线程间共享多个资源的时候，如果两个线程分别占有一部分资源并且同时等待对方的资源，就会造成死锁。

尽管死锁很少发生，但一旦发生就会造成应用的停止响应。下面看一个死锁的例子

""" Python 死锁演示 """
    import time
    import threading
    
    
    mutexA = threading.Lock()
    mutexB = threading.Lock()
    
    
    class MyThread1(threading.Thread):
    
        def run(self):
    
            # 对mutexA上锁
            mutexA.acquire()
    
            # mutexA上锁后，延时1秒，等待另外那个线程 把mutexB上锁
            print(self.name+'----do1---up----')
            time.sleep(1)
    
            # 此时会堵塞，因为这个mutexB已经被另外的线程抢先上锁了
            mutexB.acquire()
            print(self.name+'----do1---down----')
            mutexB.release()
    
            # 对mutexA解锁
            mutexA.release()
    
    
    class MyThread2(threading.Thread):
    
        def run(self):
            # 对mutexB上锁
            mutexB.acquire()
    
            # mutexB上锁后，延时1秒，等待另外那个线程 把mutexA上锁
            print(self.name+'----do2---up----')
            time.sleep(1)
    
            # 此时会堵塞，因为这个mutexA已经被另外的线程抢先上锁了
            mutexA.acquire()
            print(self.name+'----do2---down----')
            mutexA.release()
    
            # 对mutexB解锁
            mutexB.release()
    
    
    def main():
        t1 = MyThread1()
        t2 = MyThread2()
    
        t1.start()
        t2.start()
    
    
    if __name__ == '__main__':
        main()

运行结果：

Thread-1----do1---up----
    Thread-2----do2---up----

此时已经进入到了死锁状态，可以使用 `ctrl-c` 退出

**避免死锁**

*  程序设计时要尽量避免死锁（银行家算法）
 *  添加超时时间等

## 银行家算法 ##

### 背景知识 ###

一个银行家如何将一定数目的资金安全地借给若干个客户，使这些客户既能借到钱完成要干的事，同时银行家又能收回全部资金而不至于破产，这就是银行家问题。这个问题同操作系统中资源分配问题十分相似：银行家就像一个操作系统，客户就像运行的进程，银行家的资金就是系统的资源。

### 问题的描述 ###

一个银行家拥有一定数量的资金，有若干个客户要贷款。每个客户须在一开始就声明他所需贷款的总额。若该客户贷款总额不超过银行家的资金总数，银行家可以接收客户的要求。客户贷款是以每次一个资金单位（如1万RMB等）的方式进行的，客户在借满所需的全部单位款额之前可能会等待，但银行家须保证这种等待是有限的，可完成的。

例如：有三个客户C1，C2，C3，向银行家借款，该银行家的资金总额为10个资金单位，其中C1客户要借9各资金单位，C2客户要借3个资金单位，C3客户要借8个资金单位，总计20个资金单位。某一时刻的状态如图所示。

![银行卡算法][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjI5ODU3_size_16_color_FFFFFF_t_70]

对于 **a图** 的状态，按照 **安全序列** 的要求，我们选的第一个客户应满足该客户所需的贷款小于等于银行家当前所剩余的钱款，可以看出只有 **C2客户** 能被满足：**C2客户需1个资金单位**，小银行家手中的 2 个资金单位，于是银行家把 1 个资金单位借给 **C2客户**，使之完成工作并归还所借的 3 个资金单位的钱，进入 **b图**。同理，银行家把4个资金单位借给 **C3客户**，使其完成工作，在 **c图** 中，只剩一个 **客户C1**，它需 7 个资金单位，这时银行家有 8 个资金单位，所以C1也能顺利借到钱并完成工作。最后（见图d）银行家收回全部 10个资金单位，保证不赔本。那么客户序列 `{C1，C2，C3}`就是个安全序列，按照这个序列贷款，银行家才是安全的。否则的话，若在图b状态时，银行家把手中的4个资金单位借给了 `C1`，则出现不安全状态：这时 `C1，C3` 均不能完成工作，而银行家手中又没有钱了，系统陷入僵持局面，银行家也不能收回投资。

综上所述，银行家算法是从当前状态出发，逐个按安全序列检查各客户谁能完成其工作，然后假定其完成工作且归还全部贷款，再进而检查下一个能完成工作的客户，…。如果所有客户都能完成工作，则找到一个安全序列，银行家才是安全的。

## 公众号 ##

**新建文件夹X**

> 大自然用数百亿年创造出我们现实世界，而程序员用几百年创造出一个完全不同的虚拟世界。我们用键盘敲出一砖一瓦，用大脑构建一切。人们把1000视为权威，我们反其道行之，捍卫1024的地位。我们不是键盘侠，我们只是平凡世界中不凡的缔造者 。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNjI5ODU3_size_16_color_FFFFFF_t_70]: /images/20221023/7e9c72fc34ab4f9e9ab6403c0250e74b.png