1. 多线程编程与线程安全相关重要概念
在我的上篇博文 聊聊 Python 中的 GIL 中, 我们熟悉了几个特别重要的概念: GIL, 线程, 进程, 线程安全, 原子操作.
以下是简单回顾, 详细介绍请直接看聊聊 Python 中的 GIL
GIL: Global Interpreter Lock, 全局解释器锁. 为了解决多线程之间数据完整性和状态同步的问题, 设计为在任意时刻只有一个线程在解释器中运行.
线程: 程序执行的最小单位.
进程: 系统资源分配的最小单位.
线程安全: 多线程环境中, 共享数据同一时间只能有一个线程来操作.
原子操作: 原子操作就是不会因为进程并发或者线程并发而导致被中断的操作.
还有一个重要的结论: 当对全局资源存在写操作时, 如果不能保证写入过程的原子性, 会出现脏读脏写的情况, 即线程不安全. Python 的 GIL 只能保证原子操作的线程安全, 因此在多线程编程时我们需要通过加锁来保证线程安全.
最简单的锁是互斥锁(同步锁), 互斥锁是用来解决 io 密集型场景产生的计算错误, 即目的是为了保护共享的数据, 同一时间只能有一个线程来修改共享的数据.
下面我们会来介绍如何使用互斥锁.
2. Threading.Lock 实现互斥锁的简单示例
我们通过 Threading.Lock()来实现锁.
以下是线程不安全的例子:
- >>> import threading
- >>> import time
- >>> def sub1():
- global count
- tmp = count
- time.sleep(0.001)
- count = tmp + 1
- time.sleep(2)
- >>> count = 0
- >>> def verify(sub):
- global count
- thread_list = []
- for i in range(100):
- t = threading.Thread(target=sub,args=())
- t.start()
- thread_list.append(t)
- for j in thread_list:
- j.join()
- print(count)
- >>> verify(sub1)
- 14
在这个例子中, 我们把
count+=1
代替为
- tmp = count
- time.sleep(0.001)
- count = tmp + 1
是因为, 尽管 count+=1 是非原子操作, 但是因为 CPU 执行的太快了, 比较难以复现出多进程的非原子操作导致的进程不安全. 经过代替之后, 尽管只 sleep 了 0.001 秒, 但是对于 CPU 的时间来说是非常长的, 会导致这个代码块执行到一半, GIL 锁就释放了. 即 tmp 已经获取到 count 的值了, 但是还没有将 tmp + 1 赋值给 count. 而此时其他线程如果执行完了 count = tmp + 1, 当返回到原来的线程执行时, 尽管 count 的值已经更新了, 但是 count = tmp + 1 是个赋值操作, 赋值的结果跟 count 的更新的值是一样的. 最终导致了我们累加的值有很多丢失.
下面是线程安全的例子, 我们可以用 threading.Lock()获得锁
- >>> count = 0
- >>> def sub2():
- global count
- if lock.acquire(1):
- #acquire()是获取锁, acquire(1)返回获取锁的结果, 成功获取到互斥锁为 True, 如果没有获取到互斥锁则返回 False
- tmp = count
- time.sleep(0.001)
- count = tmp + 1
- time.sleep(2)
lock.release() 一系列操作结束之后需要释放锁
- >>> def verify(sub):
- global count
- thread_list = []
- for i in range(100):
- t = threading.Thread(target=sub,args=())
- t.start()
- thread_list.append(t)
- for j in thread_list:
- j.join()
- print(count)
- >>> verify(sub2)
- 100
获取锁和释放锁的语句也可以用 Python 的 with 来实现, 这样更简洁.
- >>> count = 0
- >>> def sub3():
- global count
- with lock:
- tmp = count
- time.sleep(0.001)
- count = tmp + 1
- time.sleep(2)
- >>> def verify(sub):
- global count
- thread_list = []
- for i in range(100):
- t = threading.Thread(target=sub,args=())
- t.start()
- thread_list.append(t)
- for j in thread_list:
- j.join()
- print(count)
- >>> verify(sub3)
- 100
3. 两种死锁情况及处理
死锁产生的原因
两种死锁:
3.1 迭代死锁与递归锁(RLock)
该情况是一个线程 "迭代" 请求同一个资源, 直接就会造成死锁. 这种死锁产生的原因是我们标准互斥锁 threading.Lock 的缺点导致的. 标准的锁对象 (threading.Lock) 并不关心当前是哪个线程占有了该锁; 如果该锁已经被占有了, 那么任何其它尝试获取该锁的线程都会被阻塞, 包括已经占有该锁的线程也会被阻塞.
下面是例子,
- #/usr/bin/python3
- # -*- coding: utf-8 -*-
- import threading
- import time
- count_list = [0,0]
- lock = threading.Lock()
- def change_0():
- global count_list
- with lock:
- tmp = count_list[0]
- time.sleep(0.001)
- count_list[0] = tmp + 1
- time.sleep(2)
- print("Done. count_list[0]:%s" % count_list[0])
- def change_1():
- global count_list
- with lock:
- tmp = count_list[1]
- time.sleep(0.001)
- count_list[1] = tmp + 1
- time.sleep(2)
- print("Done. count_list[1]:%s" % count_list[1])
- def change():
- with lock:
- change_0()
- time.sleep(0.001)
- change_1()
- def verify(sub):
- global count_list
- thread_list = []
- for i in range(100):
- t = threading.Thread(target=sub, args=())
- t.start()
- thread_list.append(t)
- for j in thread_list:
- j.join()
- print(count_list)
- if __name__ == "__main__":
- verify(change)
示例中, 我们有一个共享资源 count_list, 有两个分别取这个共享资源第一部分和第二部分的数字 (count_list[0] 和 count_list[1]). 两个访问函数都使用了锁来确保在获取数据时没有其它线程修改对应的共享数据.
现在, 如果我们思考如何添加第三个函数来获取两个部分的数据. 一个简单的方法是依次调用这两个函数, 然后返回结合的结果.
这里的问题是, 如有某个线程在两个函数调用之间修改了共享资源, 那么我们最终会得到不一致的数据.
最明显的解决方法是在这个函数中也使用 lock. 然而, 这是不可行的. 里面的两个访问函数将会阻塞, 因为外层语句已经占有了该锁.
结果是没有任何输出, 死锁.
为了解决这个问题, 我们可以用 threading.RLock 代替 threading.Lock
- #/usr/bin/python3
- # -*- coding: utf-8 -*-
- import threading
- import time
- count_list = [0,0]
- lock = threading.RLock()
- def change_0():
- global count_list
- with lock:
- tmp = count_list[0]
- time.sleep(0.001)
- count_list[0] = tmp + 1
- time.sleep(2)
- print("Done. count_list[0]:%s" % count_list[0])
- def change_1():
- global count_list
- with lock:
- tmp = count_list[1]
- time.sleep(0.001)
- count_list[1] = tmp + 1
- time.sleep(2)
- print("Done. count_list[1]:%s" % count_list[1])
- def change():
- with lock:
- change_0()
- time.sleep(0.001)
- change_1()
- def verify(sub):
- global count_list
- thread_list = []
- for i in range(100):
- t = threading.Thread(target=sub, args=())
- t.start()
- thread_list.append(t)
- for j in thread_list:
- j.join()
- print(count_list)
- if __name__ == "__main__":
- verify(change)
3.2 互相等待死锁与锁的升序使用
死锁的另外一个原因是两个进程想要获得的锁已经被对方进程获得, 只能互相等待又无法释放已经获得的锁, 而导致死锁. 假设银行系统中, 用户 a 试图转账 100 块给用户 b, 与此同时用户 b 试图转账 500 块给用户 a, 则可能产生死锁.
2 个线程互相等待对方的锁, 互相占用着资源不释放.
下面是一个互相调用导致死锁的例子:
- #/usr/bin/python3
- # -*- coding: utf-8 -*-
- import threading
- import time
- class Account(object):
- def __init__(self, name, balance, lock):
- self.name = name
- self.balance = balance
- self.lock = lock
- def withdraw(self, amount):
- self.balance -= amount
- def deposit(self, amount):
- self.balance += amount
- def transfer(from_account, to_account, amount):
- with from_account.lock:
- from_account.withdraw(amount)
- time.sleep(1)
- print("trying to get %s's lock..." % to_account.name)
- with to_account.lock:
- to_account_deposit(amount)
- print("transfer finish")
- if __name__ == "__main__":
- a = Account('a',1000, threading.Lock())
- b = Account('b',1000, threading.Lock())
- thread_list = []
- thread_list.append(threading.Thread(target = transfer, args=(a,b,100)))
- thread_list.append(threading.Thread(target = transfer, args=(b,a,500)))
- for i in thread_list:
- i.start()
- for j in thread_list:
- j.join()
最终的结果是死锁:
- trying to get account a's lock...
- trying to get account b's lock...
即我们的问题是:
你正在写一个多线程程序, 其中线程需要一次获取多个锁, 此时如何避免死锁问题.
解决方案:
在多线程程序中, 死锁问题很大一部分是由于线程同时获取多个锁造成的. 举个例子: 一个线程获取了第一个锁, 然后在获取第二个锁的 时候发生阻塞, 那么这个线程就可能阻塞其他线程的执行, 从而导致整个程序假死. 其实解决这个问题, 核心思想也特别简单: 目前我们遇到的问题是两个线程想获取到的锁, 都被对方线程拿到了, 那么我们只需要保证在这两个线程中, 获取锁的顺序保持一致就可以了. 举个例子, 我们有线程 thread_a, thread_b, 锁 lock_1, lock_2. 只要我们规定好了锁的使用顺序, 比如先用 lock_1, 再用 lock_2, 当线程 thread_a 获得 lock_1 时, 其他线程如 thread_b 就无法获得 lock_1 这个锁, 也就无法进行下一步操作(获得 lock_2 这个锁), 也就不会导致互相等待导致的死锁. 简言之, 解决死锁问题的一种方案是为程序中的每一个锁分配一个唯一的 id, 然后只允许按照升序规则来使用多个锁, 这个规则使用上下文管理器 是非常容易实现的, 示例如下:
- #/usr/bin/python3
- # -*- coding: utf-8 -*-
- import threading
- import time
- from contextlib import contextmanager
- thread_local = threading.local()
- @contextmanager
- def acquire(*locks):
- #sort locks by object identifier
- locks = sorted(locks, key=lambda x: id(x))
- #make sure lock order of previously acquired locks is not violated
- acquired = getattr(thread_local,'acquired',[])
- if acquired and (max(id(lock) for lock in acquired)>= id(locks[0])):
- raise RuntimeError('Lock Order Violation')
- # Acquire all the locks
- acquired.extend(locks)
- thread_local.acquired = acquired
- try:
- for lock in locks:
- lock.acquire()
- yield
- finally:
- for lock in reversed(locks):
- lock.release()
- del acquired[-len(locks):]
- class Account(object):
- def __init__(self, name, balance, lock):
- self.name = name
- self.balance = balance
- self.lock = lock
- def withdraw(self, amount):
- self.balance -= amount
- def deposit(self, amount):
- self.balance += amount
- def transfer(from_account, to_account, amount):
- print("%s transfer..." % amount)
- with acquire(from_account.lock, to_account.lock):
- from_account.withdraw(amount)
- time.sleep(1)
- to_account.deposit(amount)
- print("%s transfer... %s:%s ,%s: %s" % (amount,from_account.name,from_account.balance,to_account.name, to_account.balance))
- print("transfer finish")
- if __name__ == "__main__":
- a = Account('a',1000, threading.Lock())
- b = Account('b',1000, threading.Lock())
- thread_list = []
- thread_list.append(threading.Thread(target = transfer, args=(a,b,100)))
- thread_list.append(threading.Thread(target = transfer, args=(b,a,500)))
- for i in thread_list:
- i.start()
- for j in thread_list:
- j.join()
我们获得的结果是
- 100 transfer...
- 500 transfer...
- 100 transfer... a:900 ,b:1100
- transfer finish
- 500 transfer... b:600, a:1400
- transfer finish
成功的避免了互相等待导致的死锁问题.
在上述代码中, 有几点语法需要解释:
1. 装饰器 @contextmanager 是用来让我们能用 with 语句调用锁的, 从而简化锁的获取和释放过程. 关于 with 语句, 大家可以参考浅谈 Python 的 with 语句 (https://www.ibm.com/developerworks/cn/opensource/os-cn-pythonwith/). 简言之, with 语句在调用时, 先执行 __enter__() 方法, 然后执行 with 结构体内的语句, 最后执行__exit__()语句. 有了装饰器 @contextmanager. 生成器函数中 yield 之前的语句在 __enter__() 方法中执行, yield 之后的语句在 __exit__() 中执行, 而 yield 产生的值赋给了 as 子句中的 value 变量.
2. try 和 finally 语句中实现的是锁的获取和释放.
3. try 之前的语句, 实现的是对锁的排序, 以及锁排序是否被破坏的判断.
今天我们主要讨论了 Python 多线程中如何保证线程安全, 互斥锁的使用方法. 另外着重讨论了两种导致死锁的情况: 迭代死锁与互相等待死锁, 以及这两种死锁的解决方案: 递归锁 (RLock) 的使用和锁的升序使用.
对于多线程编程, 我们将在下一篇文章讨论线程同步 (Event) 问题, 以及对 Python 多线程模块 (threading) 进行总结.
来源: https://www.cnblogs.com/ArsenalfanInECNU/p/10022740.html