多并发编程-多线程#

守护线程/守护进程 - daemon#

如果主线程结束了守护线程也不再保留即使守护线程还在执行（没有结束）

1
import threading
2
import time
3

4

5
def show_message(content):
6
    while True:
7
        print(content, end='')
8

9

10
def main():
11
    threading.Thread(
12
        target=show_message, args=('Ping', ), daemon=True
13
    ).start()
14
    threading.Thread(
15
        target=show_message, args=('Pong', ), daemon=True
16
    ).start()
17
    sleep(5)

上面这段直接跑会报 NameError: name 'sleep' is not defined——sleep 来自 time 模块，要么 import time 后写 time.sleep(5)，要么 from time import sleep。

Lock - 多个线程竞争临界资源（资源只有1个） - 获得锁才能操作资源

Condition — 基于Lock对象可以创建它 - wait() / notify_all()

实现线程调度

Semaphore - 多个线程竞争资源（资源有多个，但线程数量大于资源数量）

多个线程通信比较简单因为可以共享内存
多个进程通信相对比较困难，可以使用multiprocessing.Queue，通过多个进程共享一个队列来实现进程间的通信

生产者消费者模型 / 哲学家进餐模型 - 多线程编程模型

三种方式实现多线程#

多线程程序如果没有竞争资源的场景那么通常会比较简单

临界资源 - 被多个线程竞争的资源

当多个线程竞争临界资源的时候如果缺乏必要的保护措施就会导致数据错乱

1. 定义线程类#

1
import threading
2
import time
3

4

5
class Account(object):
6
    """银行账户"""
7

8
    def __init__(self):
9
        self.balance = 0.0
10
        self.lock = threading.Lock()
11

12
    def deposit(self, money):
13
        # 通过锁保护临界资源
14
        # 可以写try-finally也可以使用上下文语法
15
        # self.lock.acquire()
16
        # try:
17
        #     pass
18
        # finally:
19
        #     self.lock.release()
20
        with self.lock:
21
            new_balance = self.balance + money
22
            time.sleep(0.001)
23
            self.balance = new_balance
24

25

26
class AddMoneyThread(threading.Thread):
27

28
    def __init__(self, account, money):
29
        self.account = account
30
        self.money = money
31
        # 自定义线程的初始化方法中必须调用父类的初始化方法
32
        super().__init__()
33

34
    # run方法是线程启动之后要执行的回调方法（钩子函数）
35
    # 所以启动线程不能够直接调用run方法而是通过start方法启动线程
36
    # 什么时候需要使用回调式编程？
37
    # 你知道要做什么但不知道什么时候会做这件事情
38
    def run(self):
39
        # 线程启动之后要执行的操作
40
        self.account.deposit(self.money)
41

42

43
def main():
44
    a1 = Account()
45
    threads = []
46
    for _ in range(100):
47
        t = AddMoneyThread(a1, 1)
48
        threads.append(t)
49
        t.start()
50
    for t in threads:
51
        # 等待线程执行完
52
        t.join()
53

54
    print(a1.balance)
55

56

57
if __name__ == '__main__':
58
    main()

2. 直接通过threading.Thread#

1
def add_money(account, money):
2
    account.deposit(money)
3

4

5
threads = []
6
account = Account()
7
for _ in range(100):
8
    t = threading.Thread(target=add_money, args=(account, 1))
9
    threads.append(t)
10
    t.start()
11
for t in threads:
12
    t.join()
13
print(account.balance)

3.调用线程池中的线程执行特定任务#

1
from concurrent.futures import ThreadPoolExecutor
2

3

4
# 创建线程池
5
pool = ThreadPoolExecutor(max_workers=10)
6
futures = []
7
# 调用线程池中的线程来执行特定的任务
8
for _ in range(100):
9
    future = pool.submit(add_money, account, 1)
10
    futures.append(future)
11
pool.shutdown()
12
for future in futures:
13
    future.result()
14
print(account.balance)
15

16
# 也可以用上下文方法
17
futures = []
18
with ThreadPoolExecutor(max_workers=10) as pool:
19
    for _ in range(100):
20
        futures.append(pool.submit(add_money, account, 1))
21
    for future in futures:
22
        # 获取函数add_money的返回值
23
        future.result()