多线程是个坑爹的课题，有多线程就有锁。Python中有低级线程模块_thread、再封装一下就是threading、再来一下就是from concurrent.futures import ThreadPoolExecutor。其中涉及到的锁大概有Lock、RLock、Condition、Semaphore、BoundedSeamphore、Event。这么多看起来怪吓人的，其实这么多的锁全部只是由Lock演化出来的

Lock

Lock = _thread._allocate_lock 这个是直接由最低级别的_thread直接引用过来的。创建锁的时候是处于未被锁定的状态。这应该是最容易被理解和使用的一种锁了。刚学习的时候铁定写过类似下面这种示例。本质原因是python中的原子操作是针对单条指令。而a+=1被翻译成了多条指令。执行过程中任何时刻可能被打断

1
2
3
4
5
6
7
8
9
10
11
a = 0

def change():
    global a
    for i in range(100000):
        a += 1

t = [threading.Thread(target=change) for i in range(100)]
[i.start() for i in t]
[i.join() for i in t]
print(a)

解决方法就是所有线程共享同一个锁，虽然锁有很多种，但是几乎都是这样暴露给用户使用的，这是大家都知道的。一个锁都好理解。如果有两个锁就不是那么好理解了，比如这个有点装逼的例子。交叉打印Hello和World

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
lock_a = Lock()
lock_b = Lock()

def hello():
    for i in range(5):
        lock_a.acquire()
        time.sleep(0.1)
        print("Hello")
        lock_b.release()

def world():
    for i in range(5):
        lock_b.acquire()
        time.sleep(0.1)
        print("World")
        lock_a.release()

Thread(target=hello).start()
Thread(target=world).start()

RLock

可重入锁允许线程获得锁之后该线程可以无数次的再次获得锁。看起来没什么用，大概也确实没啥用吧，不过考虑一下下面这种场景

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
class Change:
    def __init__(self):
        self.a = 1
        self.b = 1
        self.lock = RLock()

    def adda(self):
        with self.lock:
            self.a += 1

    def addb(self):
        with self.lock:
            self.b += 1

    def addab(self):
        with self.lock:
            self.adda()
            self.addb()

你有一个类。其中adda改变a,addb改变b.还提供方法addab同时调用前两者。如果没有可重入锁那么你将不得不把两个函数粘贴到addab下面。它的实现原理是获取锁的时候根据get_ident得到当前线程标识。如果和当前锁的标识一样则仅仅是给值加1，否则就尝试去获取锁。简要代码如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
from threading import Lock, get_ident

class RLock:

    def __init__(self):
        self.lock = Lock()
        self._owner = None
        self._count = 0

    def acquire(self):
        me = get_ident()
        if self._owner == me:
            self._count += 1
            return
        rc = self.lock.acquire()
        if rc:
            self._owner = me
            self._count = 1

    __enter__ = acquire

    def release(self):
        if self._owner != get_ident():
            raise RuntimeError()
        self._count -= 1
        if self._count == 0:
            self._owner = None
            self.lock.release()

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.release()

Condition

被称为条件变量，这个应该是很少被使用的。它允许传入一个锁，当满足条件的时候触发通知(可以看一下queue模块的实现，当get不到数据的时候就执行wait等待，新加入数据的时候执行notify通知唤醒)，这应该是它被称为条件变量的原因(感觉都有点牵强😕),和Lock以及RLock不同。这个东东主要是用来通知其他线程的。大概有两个方法wait(阻塞直到被通知)，notify(唤醒wait)。
实现的思路是所有的线程共享同一个锁A，同时有一个共用的锁列表。每当调用wait的时候生成一个新的锁放入锁列表，然后获得该锁，再释放锁A(因为wait必须在获得锁A才能调用，如果不释放则其他线程无法获得锁A)。最绝的来了，再次获得新生成的锁造成死锁。由此该线程被挂起。直到其他获得锁A的线程执行notify操作。将wait的锁释放。
实现代码如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
from collections import deque

class Condition:

    def __init__(self):
        self._lock = RLock()
        self._waiters = deque()

    def __enter__(self):
        return self._lock.__enter__()

    def __exit__(self, *args):
        return self._lock.__exit__(*args)

    def wait(self):
        assert get_ident() == self._lock._owner
        waiter = Lock()
        self._waiters.append(waiter)

        waiter.acquire()
        self._lock.release()
        waiter.acquire()
        self._lock.acquire()
        del waiter

    def notify(self):
        assert get_ident() == self._lock._owner
        for waiter in self._waiters.copy():
            waiter.release()
            self._waiters.remove(waiter)

这个东东吧其实不太适合被直接使用。但是网上还是有人硬生生凑了一个消费者生产者的例子出来做演示

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
import threading
import random
import time

class Producer(threading.Thread):
    def __init__(self, integers, condition):
        super(Producer, self).__init__()
        self.integers = integers
        self.condition = condition

    def run(self):
        for i in range(10):
            integer = random.randint(0, 256)
            with self.condition:
                self.integers.append(integer)
                self.condition.notify()
                time.sleep(0.5)

class Consumer(threading.Thread):

    def __init__(self, integers, condition):
        super(Consumer, self).__init__()
        self.integers = integers
        self.condition = condition

    def run(self):
        with self.condition:
            while True:
                while self.integers:
                    integer = self.integers.pop()
                    print(integer)
                self.condition.wait()

integers = []
condition = Condition()
Producer(integers, condition).start()
Consumer(integers, condition).start()

这个例子有个地方需要注意。notify和wait并不是一一对应的。不代表执行了notify，wait部分一定会被执行。所以如果消费者部分这样写会造成部分没有被处理

1
2
3
4
5
6
7
8
9
10
11
12
13
class Consumer(threading.Thread):

    def __init__(self, integers, condition):
        super(Consumer, self).__init__()
        self.integers = integers
        self.condition = condition

    def run(self):
        with self.condition:
            while True:
                self.condition.wait()
                integer = self.integers.pop()
                print(integer)

Semaphore & BoundedSeamphore

上面先介绍Condition是有原因的，因为多线程信号量和事件都是基于它生成的。信号量常用于限制对有限资源的访问。比如你有1000个线程，如果每一个线程都去创建数据库连接，那么数据库可能会崩。或者爬虫创建对网站的连接，太过凶残并不好，这个时候用信号量来处理就不错。

这里如果直接使用Lock来构建。那么一个锁同时只允许一个线程来获得显然是不达不到要求的。祭出刚才构建的Condition

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
class Semaphore:
    # value值用完了就被调用wait.只有notify能解除
    def __init__(self, value=1):
        self._lock = Condition()
        self.value = value

    def __enter__(self):
        with self._lock:
            if self.value == 0:
                self._lock.wait()
            else:
                self.value -= 1

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.value += 1
        self._lock.notify()


class BoundedSemaphore(Semaphore):

    def __init__(self, value=1):
        super(BoundedSemaphore, self).__init__(value=value)

Event

Event和Condition差不多。区别就是上面提到的，使用Condition，如果在单线程，那么当notify的时候是无法触发wait的，再调用wait线程会被阻塞。但是Event就支持在单线程使用。提供两个主要方法，set和wait。但是它并不需要先获得锁。使用好像更方便一点，实现如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
class Event:
    def __init__(self):
        self._cond = Condition()
        self._flag = False

    def set(self):
        with self._cond:
            self._flag = True
            self._cond.notify()

    def clear(self):
        with self._cond:
            self._flag = False

    def wait(self):
        with self._cond:
            signaled = self._flag
            if not signaled:
                signaled = self._cond.wait()
            return signaled

如果把刚才那个生产者消费者换到它上面来，大概是这个样子了

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
import threading
import random
import time

class Producer(threading.Thread):
    def __init__(self, integers, condition):
        super(Producer, self).__init__()
        self.integers = integers
        self.condition = condition

    def run(self):
        for i in range(10):
            integer = random.randint(0, 256)
            self.integers.append(integer)
            self.condition.set()
            time.sleep(0.5)

class Consumer(threading.Thread):

    def __init__(self, integers, condition):
        super(Consumer, self).__init__()
        self.integers = integers
        self.condition = condition

    def run(self):
        while True:
            while self.integers:
                integer = self.integers.pop()
                print(integer)
            self.condition.wait()

integers = []
condition = Event()
Producer(integers, condition).start()
Consumer(integers, condition).start()

当然，没啥必要用这些比较低级的东东。能用通用高级数据结构就直接用，这样大家都比较好理解，总结就是只是给共享资源加锁用Lock、限制资源访问使用BoundedSeamphore，唤醒相关线程使用Event