律回

公告

本站已于 2026.06.16 完成域名和框架迁移 >w<

Learn More

标签

律回

公告

本站已于 2026.06.16 完成域名和框架迁移 >w<

Learn More

标签

律回

公告

本站已于 2026.06.16 完成域名和框架迁移 >w<

Learn More

标签

1081 字

3 分钟

安全异步读写原理与实现

2023-09-01

computer-science

python

/

asyncio

/

async

/

concurrency

安全异步读写原理与实现#

1、基本原理#

异步读写问题与常规异步问题不同，不是单纯的互斥问题。对于安全的异步读写，为了同时保证安全性和并发性能，需要实现：

读取时：不允许写入。但可以再并行/并发地读取。
写入时：不允许读取。也不允许其他写入。

结合操作系统原理课的知识，我们可以得到用 PV 原语表示的，并发读写的伪代码：

1
# 写信号量
2
Semaphore WS = 1
3
# 当前正在读的计数变量
4
int RN = 0
5
# 变量 RN 的操作锁
6
Mutex L
7

8

9
def safe_write():
10
    P(WS)
11
    # 写操作
12
    write()
13
    V(WS)
14

15

16
def safe_read():
17
    P(L)
18
        if RN == 0:
19
            P(WS)
20
        RN += 1
21
    V(L)
22
    # 读操作
23
    read()
24
    P(L)
25
        RN -= 1
26
        if RN == 0:
27
            V(WS)
28
    V(L)

当然，在某些情况下，可能需要限制读的并发度，那么可以再添加一个信号量变量：

1
# 写信号量
2
Semaphore WS = 1
3
# 读信号量，值为一个常量
4
Semaphore RS = N
5
# 当前正在读的计数变量
6
int RN = 0
7
# 变量 RN 的操作锁
8
Mutex L
9

10

11
def safe_write():
12
    P(WS)
13
    # 写操作...
14
    write()
15
    V(WS)
16

17

18
def safe_read():
19
    P(RS)
20
        P(L)
21
            if RN == 0:
22
                P(WS)
23
            RN += 1
24
        V(L)
25
        # 读操作...
26
        read()
27
        P(L)
28
            RN -= 1
29
            if RN == 0:
30
                V(WS)
31
        V(L)
32
    V(RS)

2、代码实现#

以下所有实现，均使用 Python。

在 Python 中，PV 原语有线程版、协程版、进程版的对应实现。因此可以在各种情景下很方便地做到安全异步读写。

不过需要注意的是，由于 Python GIL 的存在，任何时候只有一个线程在运行。因此如果使用多线程，读无法实现并行，最多实现并发。当然如果使用多进程，就可以并行了。以下是多线程异步安全读写的实现示例：

1
import threading
2
import random
3
import time
4

5
write_semaphore = threading.Semaphore(1)
6
# 设置读的并发度为 2
7
read_semaphore = threading.Semaphore(2)
8
read_num = 0
9
read_num_lock = threading.Lock()
10
test_var = "initial value"
11

12
def safe_write():
13
    def _write():
14
        global test_var
15
        test_var =  random.randint(1, 1000000)
16
        # 模拟写操作需要的时间
17
        time.sleep(0.5)
18
        print(f"Changed to {test_var}")
19
    write_semaphore.acquire()
20
    _write()
21
    write_semaphore.release()
22

23
def safe_read():
24
    def _read():
25
        global test_var
26
        print(test_var)
27
    global read_num
28
    read_semaphore.acquire()
29
    with read_num_lock:
30
        if read_num == 0:
31
            write_semaphore.acquire()
32
        read_num += 1
33
    _read()
34
    with read_num_lock:
35
        read_num -= 1
36
        if read_num == 0:
37
            write_semaphore.release()
38
    read_semaphore.release()
39

40
if __name__ == "__main__":
41
    threads = []
42
    for i in range(5):
43
        threads.append(threading.Thread(target=safe_read))
44
        threads.append(threading.Thread(target=safe_write))
45
    # 打乱列表，模拟读写线程的先后到来
46
    random.shuffle(threads)
47
    for t in threads:
48
        t.start()
49
    for t in threads:
50
        t.join()

运行上述代码，获得的结果比较随机。比如：

1
Changed to 556931
2
Changed to 938396
3
938396
4
Changed to 925837
5
Changed to 436704
6
Changed to 608686
7
608686
8
608686608686
9

10
608686

简单分析一下：

读线程打印的值不会和 “Changed…” 重合，说明读写互斥
读线程打印的值会重合，且最多重合两次，说明可以并发执行，且并发度最大为 2

符合我们预设的目标。

如果使用 asyncio ，则可以实现多协程的版本，代码很类似：

1
import asyncio
2
import random
3

4
write_semaphore = asyncio.Semaphore(1)
5
read_semaphore = asyncio.Semaphore(2)
6
read_num = 0
7
read_num_lock = asyncio.Lock()
8
test_var = "initial value"
9

10
async def safe_write():
11
    async def _write():
12
        global test_var
13
        test_var =  random.randint(1, 1000000)
14
        await asyncio.sleep(0.5)
15
        print(f"Changed to {test_var}")
16
    await write_semaphore.acquire()
17
    await _write()
18
    write_semaphore.release()
19

20
async def safe_read():
21
    def _read():
22
        global test_var
23
        print(test_var)
24
    global read_num
25
    await read_semaphore.acquire()
26
    async with read_num_lock:
27
        if read_num == 0:
28
            await write_semaphore.acquire()
29
        read_num += 1
30
    _read()
31
    async with read_num_lock:
32
        read_num -= 1
33
        if read_num == 0:
34
            write_semaphore.release()
35
    read_semaphore.release()
36

37
async def main():
38
    coros = []
39
    for i in range(5):
40
        coros.append(safe_read())
41
        coros.append(safe_write())
42
    random.shuffle(coros)
43
    tasks = [asyncio.create_task(coro) for coro in coros]
44
    await asyncio.wait(tasks)
45

46
if __name__ == "__main__":
47
    asyncio.run(main())

由于 asyncio 是单线程实现的多协程，因此所有操作实际都在一个线程上。不会出现对线程不安全的 print() 的同时调用，因此打印出来的结果很整洁：

1
Changed to 17815
2
Changed to 405116
3
405116
4
405116
5
Changed to 678826
6
Changed to 88906
7
Changed to 989941
8
989941
9
989941
10
989941

3、实际使用#

在实际业务场景中，读和写的方法可能千变万化，需要按需提供。对此可以使用 Python 的上下文管理器，将实际的读写方法解耦出来：

1
import contextlib
2
import threading
3

4
write_semaphore = threading.Semaphore(1)
5
read_semaphore = threading.Semaphore(2)
6
read_num = 0
7
read_num_lock = threading.Lock()
8

9
# 如果异常没有在上下文中（即 with 子句下）捕获，则抛出
10
@contextlib.contextmanager
11
def safe_read():
12
    global read_num
13
    read_semaphore.acquire()
14
    with read_num_lock:
15
        if read_num == 0:
16
            write_semaphore.acquire()
17
        read_num += 1
18
    try:
19
        yield
20
    finally:
21
        with read_num_lock:
22
            read_num -= 1
23
            if read_num == 0:
24
                write_semaphore.release()
25
            read_semaphore.release()
26

27
@contextlib.contextmanager
28
def safe_write():
29
    write_semaphore.acquire()
30
    try:
31
        yield
32
    finally:
33
        write_semaphore.release()

使用时只需要：

1
with safe_read():
2
    ...
3
with safe_write():
4
    ...

值得注意的是，此时所有使用 safe_read、safe_write 的代码，都会共用一套读写控制逻辑，这显然不太合适。我们更希望对一个或一类资源，有独立的读写控制。这提示我们可以把它封装为类来进一步解耦：

1
import contextlib
2
import threading
3

4
class RWController:
5
    def __init__(self, max_read_num: int=None) -> None:
6
        # max_read_num 如果不为 None，意味着有读的并发限制
7
        # 注意这是 __init__() 方法下的函数
8
        @contextlib.contextmanager
9
        def safe_read():
10
            nonlocal read_num, read_semaphore, write_semaphore, read_num_lock
11
            if read_semaphore:
12
                read_semaphore.acquire()
13
            with read_num_lock:
14
                if read_num == 0:
15
                    write_semaphore.acquire()
16
                read_num += 1
17
            try:
18
                yield
19
            finally:
20
                with read_num_lock:
21
                    read_num -= 1
22
                    if read_num == 0:
23
                        write_semaphore.release()
24
                    if read_semaphore:
25
                        read_semaphore.release()
26

27
        @contextlib.contextmanager
28
        def safe_write():
29
            nonlocal write_semaphore
30
            write_semaphore.acquire()
31
            try:
32
                yield
33
            finally:
34
                write_semaphore.release()
35

36
        write_semaphore = threading.Semaphore(1)
37
        if max_read_num:
38
            read_semaphore = threading.Semaphore(max_read_num)
39
        else:
40
            read_semaphore = None
41
        read_num = 0
42
        read_num_lock = threading.Lock()
43
        self.safe_read = safe_read
44
        self.safe_write = safe_write

此时外部就可以初始化这个类，获得一个独立的读写控制器：

1
rwc1 = RWController()
2
rwc2 = RWController(max_read_num=5)
3

4
with rwc1.safe_read():
5
    ...
6
with rwc2.safe_read():
7
    ...
8
with rwc1.safe_write():
9
    ...
10
with rwc2.safe_write():
11
    ...

这样依然需要外部手动调用 with 来进行上下文管理，好处是非常灵活。

但如果只是需要某个方法获得安全的异步读写，另一个比较好的思路是通过装饰器来处理。那怎么使用装饰器处理呢？

首先让我们来看一个类的源码。它是 contextlib.ContextDecorator ：

1
class ContextDecorator(object):
2
    def _recreate_cm(self):
3
        return self
4

5
    def __call__(self, func):
6
        @wraps(func)
7
        def inner(*args, **kwds):
8
            with self._recreate_cm():
9
                return func(*args, **kwds)
10
        return inner

不难发现，继承该类并实现 __enter__() 和 __exit__() 方法后，产生的对象如果被用作装饰器方法，会执行 __call__() 这个装饰器函数。之后的执行逻辑就是：通过内部的 with 语句，在执行 __enter__() 后，执行被装饰的函数，最后再调用 __exit__() ，实现自动的上下文管理。

让我们用上这个类，来实现刚才的目标：

1
import contextlib
2
import threading
3

4
class RWController:
5
    def __init__(self, max_read_num: int=None) -> None:
6
        super().__init__()
7
        # max_read_num 不为 None，意味着有读的并发限制
8
        if max_read_num:
9
            self.read_semaphore = threading.Semaphore(max_read_num)
10
        else:
11
            self.read_semaphore = None
12
        self.write_semaphore = threading.Semaphore(1)
13
        self.read_num = 0
14
        self.read_num_lock = threading.Lock()
15
        self.safe_read = RWController.ReadController(
16
            self.read_semaphore, self.write_semaphore,
17
            self.read_num_lock, self.read_num
18
        )
19
        self.safe_write = RWController.WriteController(self.write_semaphore)
20

21
    class ReadController(contextlib.ContextDecorator):
22
        def __init__(self, read_semaphore, write_semaphore, read_num_lock, read_num) -> None:
23
            super().__init__()
24
            self.read_semaphore: threading.Semaphore = read_semaphore
25
            self.write_semaphore: threading.Semaphore = write_semaphore
26
            self.read_num_lock: threading.Lock = read_num_lock
27
            self.read_num = read_num
28

29
        def __enter__(self) -> None:
30
            # 读信号量不为 None，意味着有读的并发限制
31
            if self.read_semaphore:
32
                self.read_semaphore.acquire()
33
            with self.read_num_lock:
34
                if self.read_num == 0:
35
                    self.write_semaphore.acquire()
36
                self.read_num += 1
37

38
        def __exit__(self, *exc) -> None:
39
            with self.read_num_lock:
40
                self.read_num -= 1
41
                if self.read_num == 0:
42
                    self.write_semaphore.release()
43
                # 读信号量不为 None，意味着有读的并发限制
44
                if self.read_semaphore:
45
                    self.read_semaphore.release()
46

47
    class WriteController(contextlib.ContextDecorator):
48
        def __init__(self, write_semaphore) -> None:
49
            super().__init__()
50
            self.write_semaphore: threading.Semaphore = write_semaphore
51

52
        def __enter__(self) -> None:
53
            self.write_semaphore.acquire()
54

55
        def __exit__(self, *exc) -> None:
56
            self.write_semaphore.release()

使用的时候，只要这样就可以了：

1
rwc = RWController()
2

3
@rwc.safe_write
4
def custom_write():
5
    ...
6

7
@rwc.safe_read
8
def custom_read():
9
    ...

同理，你也可以使用多进程、多协程来实现上面的这些具体方案。对于多协程实现，需要注意：

contextmanager 有其对应的协程异步版本：asynccontextmanager，可以直接使用
ContextDecorator 类没有其对应的协程异步版本，需要自己实现。原理类似，不过是改用 async with，以及 __aenter__()、__aexit__()

4、后话#

第三方模块中，已经有比较成熟的安全异步读写解决方案了。例如基于 asyncio 多协程的实现：aiorwlock。

感兴趣的话，可以自行阅读源码了解。

如果这篇文章对你有帮助，欢迎分享给更多人！

安全异步读写原理与实现

https://miku.top/posts/safe-async-rw/

作者

律回

发布于

2023-09-01

许可协议

CC BY-SA 4.0

部分信息可能已经过时

ArchLinux 折腾指南&记录

C# 二周目学习 —— 异步编程

律回彼境

安全异步读写原理与实现#

1、基本原理#

2、代码实现#

3、实际使用#

4、后话#

目录