Python中的多线程编程，线程安全与锁(一)

时间：2021-09-04 03:59:02

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)进行总结。

参考文献：

1. 深入理解 GIL：如何写出高性能及线程安全的 Python 代码 http://python.jobbole.com/87743/

2. Python中的原子操作 https://www.jianshu.com/p/42060299c581

3. 详解python中的Lock与RLock https://blog.csdn.net/ybdesire/article/details/80294638

4. 深入解析Python中的线程同步方法 https://www.jb51.net/article/86599.htm

5.  Python中死锁的形成示例及死锁情况的防止 https://www.jb51.net/article/86617.htm

6.  举例讲解 Python 中的死锁、可重入锁和互斥锁 http://python.jobbole.com/82723/

7.  python基础之多线程锁机制

8. python--threading多线程总结

9. Python3入门之线程threading常用方法

10. 浅谈 Python 的 with 语句 https://www.ibm.com/developerworks/cn/opensource/os-cn-pythonwith/

相关文章