1. 多线程编程

1.1 线程和进程的区别

  • 进程： 进程在操作系统中可以独立运行是程序在执行过程中分配和管理资源的基本单位每一个进程都有一个自己的地址空间。
  • 线程：线程是进程中的一个实例作为系统（CPU）调度和分派的基本单位它与同属一个进程的其他的线程共享进程所拥有的全部资源。

操作系统上可以运行多个进程，一个进程可以有多个线程，多个线程可以被分配到不同的 CPU 核上跑，但实际上每个 CPU 核上只有一个线程，只是这个线程在不停的进行上下文的切换。

  • Python 多线程适合 I/O 操作密集型的任务（I/O 操作不占用 CPU，从硬盘、从网络、从内存读数据都算 I/O）。
  • Python 多线程不适合 CPU 密集操作型的任务，主要使用 CPU 来计算，如大量的数学计算。
  • 进程之间不需要使用 GIL 锁，因为进程是独立的，且默认是不会共享数据。

1.2 单线程

在单线程程序中可能包含多个方法，运行程序后，默认是在一个主线程里按顺序运行。

这意味着程序在任何给定的时间点只能执行一个操作，必须按顺序完成一项任务后才能进行下一项任务，无法同时进行多个任务。

例如，如果一个程序中有多个操作，如读取文件、进行计算和写入文件，在单线程环境中，它会先读取文件，完成后再进行计算，计算结束才会写入文件，依次顺序执行。

单线程程序相对简单，不存在线程间的同步和资源竞争问题，但在处理多个耗时操作时，可能会导致程序的响应性和效率较低。

举个计算时间的例子，在下述代码中，func1 和 func2 两个函数会依次执行，因为这是在单线程环境中，所以无法同时执行这两个任务。

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
import time
import threading

def exe_time(func):
    def new_func(*args, **kwargs):
        print("%s start: %s" % (func.__name__, time.strftime("%X", time.localtime())))
        back = func(*args, **kwargs)
        print("%s end: %s" % (func.__name__, time.strftime("%X", time.localtime())))
        return back

    return new_func

@exe_time
def func1():
    time.sleep(1)
    print("func1 执行方法")
    time.sleep(2)

@exe_time
def func2():
    time.sleep(1)
    print("func2 执行方法")
    time.sleep(2)

if __name__ == "__main__":
    before = time.time()
    func1()
    func2()
    after = time.time()
    print("taken time: %.3fs " % (after - before))

从执行结果可以看出全程花了 6.014s，因此 func1 和 func2 是串行执行的，func2 要等 func1 执行完才能执行。

运行结果：

1
2
3
4
5
6
7
8
9
func1 start: 15:10:18
func1 执行方法
func1 end: 15:10:21
func2 start: 15:10:21
func2 执行方法
func2 end: 15:10:24
taken time: 6.014s

Process finished with exit code 0

1.3 多线程

多线程是指在一个程序中同时运行多个线程。线程是进程中的执行单元，一个进程可以包含多个线程。

在多线程编程中，多个线程可以并发地执行，即在同一时间段内，多个线程都在推进各自的任务。这使得程序能够在同一时间内处理多个任务，从而提高程序的并发性和响应性。

例如，一个多线程的网络服务器可以同时处理多个客户端的请求，一个图形界面应用程序可以在处理用户输入的同时在后台进行数据加载等任务。

每个线程都有自己的执行上下文，包括程序计数器、栈等，但它们共享进程的内存空间和其他资源。

多线程编程可以提高 CPU 的利用率，特别是在存在大量 I/O 操作或等待的情况下，一个线程被阻塞时，其他线程可以继续执行。然而，多线程编程也带来了一些挑战，如线程同步、资源竞争和死锁等问题，需要谨慎处理以确保程序的正确性和稳定性。

1.3.1 实现方式

threading

在 Python3 标准库中，有两个模块 _thread 和 threading 可以提供多线程支持。但由于 _thread 是低级模块，很多功能还不完善，一般只会用到 threading 这个比较完善的高级模块，因此这里只讨论 threading 模块的使用。

Python 中使用线程有两种方式：函数或者用类来包装线程对象。

采用函数创建多线程语法如下：

1
threading.Thread(target=function, args[, kwargs])

参数说明:

  • function - 线程函数
  • args - 传递给线程函数的参数，他必须是个 tuple 类型
  • kwargs - 可选参数

  • 不带参数

我们还是以上面那个计算时间的函数为例，这里我们将单线程优化成多线程模式。

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
import time
import threading

def exe_time(func):
    def new_func(*args, **kwargs):
        print("%s start: %s" % (func.__name__, time.strftime("%X", time.localtime())))
        back = func(*args, **kwargs)
        print("%s end: %s" % (func.__name__, time.strftime("%X", time.localtime())))
        return back

    return new_func

@exe_time
def func1():
    time.sleep(1)
    print("func1 执行方法")
    time.sleep(2)

@exe_time
def func2():
    time.sleep(1)
    print("func2 执行方法")
    time.sleep(2)

if __name__ == "__main__":
    before = time.time()
    t1 = threading.Thread(target=func1)
    t1.start()
    t2 = threading.Thread(target=func2)
    t2.start()
    t1.join()
    t2.join()
    after = time.time()
    print("taken time: %.3fs " % (after - before))

可以发现执行时间明显要快了差不多一倍。

运行结果：

1
2
3
4
5
6
7
8
9
func1 start:15:17:46
func2 start:15:17:46
func1 执行方法
func2 执行方法
func1 end:15:17:49
func2 end:15:17:49
taken time:3.009s

Process finished with exit code 0
  • 带参数

同样，我们再举一个带参数的例子。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import threading
import time

# 定义线程执行的函数
def thread_function(name):
    print(f"Thread {name} is running")
    time.sleep(2)
    print(f"Thread {name} completed")

# 创建线程
thread1 = threading.Thread(target=thread_function, args=("Thread 1",))
thread2 = threading.Thread(target=thread_function, args=("Thread 2",))

# 启动线程
thread1.start()
thread2.start()

# 等待线程完成（可选）
thread1.join()
thread2.join()

运行结果：

1
2
3
4
5
Thread Thread 1 is running
Thread Thread 2 is running
Thread Thread 1 completedThread Thread 2 completed

Process finished with exit code 0
继承 Thread 类

我们也可以通过直接从 threading.Thread 继承创建一个新的子类，并实例化后调用 start() 方法启动新线程，即它调用了线程的 run() 方法。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import threading
import time

class MyThread(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)
    def run(self):
        #重写 run 方法，run 方法中是该线程要执行的动作
        print("开始线程...")
        time.sleep(3)
        print("退出线程...")

# 创建自定义的线程类对象
t1 = MyThread()

# 调用自定义线程类中的继承自父类的 start 方法,执行自定义线程类中的 run 方法
t1.start()
t1.join()
print('ending')

运行结果：

1
2
3
4
5
开始线程...
退出线程...
ending

Process finished with exit code 0

1.3.2 常用方法及属性

方法/属性：t = threading.current_thread() 作用
t.start() 启动线程
t.name 获取或设置线程的名称
t.name = xxx 设置线程的名称为 xxx
t.is_alive() 判断线程是否为激活状态
t.daemon = True 可以通过设置线程的 daemon 属性为 True 来将其设置为守护线程
t.isDaemon() 判断是否为守护线程
t.join() 逐个执行每个线程，执行完毕后继续往下执行，该方法使得多线程变得无意义

程序默认在一个线程中执行，这个线程称为主线程。

1
2
3
4
5
import threading

t = threading.current_thread()
print(t.name)           # MainThread
print(t.is_alive())     # True
join 方法

join() 方法：主线程 A 中，创建了子线程 B，并且在主线程 A 中调用了 B.join()。那么主线程 A 会在调用的地方等待，直到子线程 B 完成操作后，才可以接着往下执行。

语法：

1
join([timeout])

参数为可选，代表线程运行的最大时间，即如果超过这个时间，不管这个此线程有没有执行完毕都会被回收，然后主线程或函数都会接着执行的。

  • 不带 timeout 参数

先举个不带参数的例子，可以发现使用 t.join() 之后，主线程会等待线程 t 执行完之后，才会执行后续的逻辑。

1
2
3
4
5
6
7
8
9
10
11
12
import threading
import time

def search():
    print('start search...')
    time.sleep(5)
    print("end search...")

t = threading.Thread(target=search)
t.start()
t.join()
print("helle world 1")

运行结果：

1
2
3
4
5
start search..
end search....
helle world1

Process finished with exit code 0
  • 带 timeout 参数

再来看一个带参数的例子，我这里设置了 1 秒的超时时间即 t.join(1)，而这 1 秒的时间要比我子线程执行的时间要短。因此，主线程只会等 1 秒钟，不会等到子线程 t 执行完，而是继续执行后面的打印操作。

1
2
3
4
5
6
7
8
9
10
11
12
import threading
import time

def search():
    print('start search...')
    time.sleep(5)
    print("end search...")

t = threading.Thread(target=search)
t.start()
t.join(1)    # 设置1秒钟超时时间
print("helle world 1")

运行结果：

1
2
3
4
5
start search..
helle world 1
end search....

Process finished with exit code 0
daemon 属性

daemon 属性用于将线程设置为守护线程（Daemon Thread）。

守护线程是一种在后台运行的线程，当所有的非守护线程结束时，守护线程会自动结束，即使它仍在运行中。

Tips：

daemon 属性必须在 t.start() 即子线程执行之前设置，不然不起作用。

  • 不设置 daemon 属性

先来看一下下述案例正常执行的效果，可以发现主线程并没有等待子线程执行完，就往下执行了。

1
2
3
4
5
6
7
8
9
10
11
import threading
import time

def search():
    for i in range(3):
        time.sleep(1)
        print("helle world 2")

t = threading.Thread(target=search)
t.start()
print("helle world 1")

运行结果：

1
2
3
4
5
6
helle world 1
helle world 2
helle world 2
helle world 2

Process finished with exit code 0
  • 设置 daemon 属性

接下来，我将案例中的子线程的 daemon 属性设置为 True，即设置为守护线程。那么主线程执行结束之后，子线程也会跟着结束，不会执行后续的操作了。

1
2
3
4
5
6
7
8
9
10
11
12
import threading
import time

def search():
    for i in range(3):
        time.sleep(1)
        print("helle world 2")

t = threading.Thread(target=search)
t.daemon = True
t.start()
print("helle world 1")

运行结果：

1
2
3
helle world 1

Process finished with exit code 0

上面这个结果，看不出子线程是否执行了。为了理解的更透彻一些，我在主线程后再加两行代码，主线程执行完第一个 print 语句后不会立马结束，而是睡眠个 2 秒，再执行一个 print 语句才结束。

那么从结果中就可以发现，子线程确实执行了的，因为打印了一行语句出来，只不过在主线程结束之后，就没有继续往下执行了。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
import threading
import time

def search():
    for i in range(3):
        time.sleep(1)
        print("helle world 2")

t = threading.Thread(target=search)
t.daemon = True
t.start()
print("helle world 1")
time.sleep(2)
print("end...")

运行结果：

1
2
3
4
5
helle world 1
helle world 2
end...

Process finished with exit code 0

2. 多进程编程

2.1 定义

multiprocessing 是 python 的多进程管理包，和 threading.Thread 类似。

multiprocessing 模块可以让程序员在给定的机器上充分的利用 CPU。在 multiprocessing 中，通过创建 Process 对象生成进程，然后调用它的 start() 方法。

2.1.1 关键概念和特点

  1. Process 类：用于创建进程。可以通过指定目标函数和传递参数来创建新的进程。

  2. 进程间通信：

    • 可以使用 Queue 实现进程间的消息传递。
    • Pipe 用于两个进程之间的双向通信。

  3. 共享数据：

    • 可以使用 Value 和 Array 来在进程间共享简单的数据类型。

  4. 进程池 Pool：

    • 方便管理和复用进程，适用于需要创建多个相同任务的进程的情况。

2.1.2 优缺点

  • 优点：
    • 真正的并行执行：因为每个进程都有独立的内存空间和 GIL，所以可以在多核 CPU 上同时运行，适用于 CPU 密集型任务。
    • 资源隔离：每个进程有独立的资源，一个进程的错误或崩溃通常不会影响其他进程。
  • 缺点：
    • 进程创建开销大：创建新进程比创建新线程的系统开销更大。
    • 数据共享复杂：进程间数据共享不如线程间方便，需要特殊的机制来实现。

2.2 实现方式

2.2.1 创建多进程语法

  • function - 函数。
  • args - 传递给函数的参数，他必须是个 tuple 类型。
  • kwargs - 可选参数。
1
multiprocessing.Process (function, args[, kwargs])

2.2.2 多进程示例

多进程和多线程实现的方式和比较类似，只是调用的函数方法不同。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import multiprocessing
import time

def worker(name):
    print(f"{name} 开始工作")
    time.sleep(2)
    print(f"{name} 工作完成")

if __name__ == '__main__':
    # 创建进程
    process1 = multiprocessing.Process(target=worker, args=("进程 1",))
    process2 = multiprocessing.Process(target=worker, args=("进程 2",))

    # 启动进程
    process1.start()
    process2.start()

    # 等待进程完成
    process1.join()
    process2.join()

    print("所有进程完成")

2.3 常用方法及属性

Process([group [, target [, name [, args [, kwargs]]]]]) 作用
target 如果传递了函数的引用，可以认为这个子进程就执行这里的代码
args 给 target 指定的函数传递的参数，以元组的方式传递
kwargs 给 target 指定的函数传递命名参数
start() 启动子进程实例（创建子进程）
is_alive() 判断进程子进程是否还在活着
join([timeout]) 是否等待子进程执行结束，或等待多少秒
terminate() 不管任务是否完成，立即终止子进程
name 当前进程的别名，默认为 Process-N，N 为从 1 开递增的整数
pid 当前进程的 pid（进程号）

2.4 进程间通信

2.4.1 Queue

在 Python 中，进程各自持有一份数据，默认无法共享数据，可以使用 multiprocessing 模块的 Queue 实现多进程之间的数据传递，Queue 本身是一个消息列队程序。

我们这里就举一个比较经典的生产者消费者的例子，这里我们创建了一个 Queue 对象 q，producer 函数向队列中放入数据，consumer 函数从队列中取出数据。

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
import multiprocessing

# 生产者方法
def producer(q):
    for i in range(5):
        q.put(i)
        print(f"生产者放入: {i}")

# 消费者方法
def consumer(q):
    while True:
        if not q.empty():
            item = q.get()
            print(f"消费者取出: {item}")
        else:
            break

if __name__ == "__main__":
    # 创建队列
    q = multiprocessing.Queue()

    # 创建生产者进程
    p1 = multiprocessing.Process(target=producer, args=(q,))
    # 创建消费者进程
    p2 = multiprocessing.Process(target=consumer, args=(q,))

    # 启动进程
    p1.start()
    p2.start()

    # 等待进程结束
    p1.join()
    p2.join()

我们再看一个例子帮助大家理解更深刻一些，下面来模拟一下数据下载和数据处理的逻辑，同样可以针对这两个逻辑创建两个进程来执行，然后通过一个队列来进行进程间的通信。

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
38
39
40
41
42
43
44
45
import multiprocessing
import time

def download(q):
    """模拟下载数据的功能"""
    data = [11, 22, 33, 44]

    # 向队列中写入数据
    for temp in data:
        q.put(temp)

    print("----下载器已经下载完毕数据,并将数据存入到队列中----")

def analysis(q):
    """"数据处理"""
    analysis_data = list()
    time.sleep(1)

    # 从队列中获取数据
    while True:
        data = q.get()
        print(data)
        analysis_data.append(data)

        if q.empty():
            break

    # 对数据进行处理
    print(analysis_data)

if __name__ == "__main__":
    # 创建一个队列
    q = multiprocessing.Queue()

    # 创建多个进程，将队列对象作为参数传递到子进程中
    p1 = multiprocessing.Process(target=download, args=(q,))
    p2 = multiprocessing.Process(target=analysis, args=(q,))

    # 启动进程
    p1.start()
    p2.start()

    # 等待进程结束
    p1.join()
    p2.join()

2.4.2 Pool

Pool 类可以提供指定数量的进程供用户调用，当有新的请求提交到 Pool 中时，如果池还没有满，就会创建一个新的进程来执行请求。如果池满，请求就会告知先等待，直到池中有进程结束，才会创建新的进程来执行这些请求。

下面来看一个例子，在我标黄的部分可以看到有两种实现方法，如果是运行我注释掉的那行代码，则进程会是以串行的方式运行，执行的时间会非常长；但如果是使用 apply_async 方法，则进程将以并行的方式运行，并行进程最大的数量为我前面定义的 5 个进程，此时程序将会很快执行完毕。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import multiprocessing
import time
import os
from multiprocessing import Pool

def process(i):
    time.sleep(1)
    print("%s in process %s" % (multiprocessing.current_process().name, i))

def callback_fun(i):
    print('%s -->exec done: %s' % (multiprocessing.current_process().name, os.getpid()))


if __name__ == "__main__":
    pool = Pool(processes=5)   # 允许进程池同时放入 5 个进程
    print("开始执行主进程", os.getpid())
    for i in range(10):
        # pool.apply(func=process,args=(i,))    # 串行
        pool.apply_async(func=process, args=(i,))     # 并行

    pool.close()
    pool.join()     # 进程池中进程执行完毕后再关闭，如果注释，那么程序直接参照
    print('main process end')

我们还可以设置回调函数，即在每个进程执行完之后，再执行一次设置的回调函数。

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
import multiprocessing
import time
import os
from multiprocessing import Pool

def process(i):
    time.sleep(1)
    print("%s in process %s" % (multiprocessing.current_process().name, i))

def callback_fun(i):
    print('%s -->exec done: %s' % (multiprocessing.current_process().name, os.getpid()))


if __name__ == "__main__":
    pool = Pool(processes=5)   # 允许进程池同时放入 5 个进程
    # 记录开始时间
    start_time = time.time()

    print("开始执行主进程", os.getpid())
    for i in range(10):
        pool.apply_async(func=process, args=(i,), callback=callback_fun)   # callback = 回调

    pool.close()
    pool.join()     # 进程池中进程执行完毕后再关闭，如果注释，那么程序直接参照
    print('main process end')

    # 记录结束时间
    end_time = time.time()
    # 计算并打印执行时间
    execution_time = end_time - start_time
    print(f"执行时间: {execution_time} 秒")

运行结果：

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
开始执行主进程89957
SpawnPooLWorker-1 in process 0
MainProcess -->exec done:89957
SpawnPooLWorker-2 in process 1
SpawnPooLWorker-3 in process 2
MainProcess -->exec done:89957
SpawnPooLWorker-5 in process 3
MainProcess -->exec done:89957
MainProcess -->exec done:89957
SpawnPooLWorker-4 in process 4
MainProcess -->exec done:89957
SpawnPooLWorker-1 in process 5
MainProcess -->exec done:89957
SpawnPooLWorker-2 in process 6
MainProcess -->exec done:89957
SpawnPooLWorker-5 in process 8
MainProcess -->exec done:89957
SpawnPooLWorker-3 in process 7
MainProcess -->exec done:89957
SpawnPooLWorker-4 in process 9
MainProcess -->exec done:89957
main process end

执行时间：2.1066787242889404秒
Process finished with exit code 0

Tips：

Pool 中创建的进程数量并不是越多越好，一般设置为与系统的 CPU 核心数相当，或者略多一些。

另外，除了使用 apply_async 方法可以实现并发操作外，我们还可以使用 map 方法来实现。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
import time
from multiprocessing import Pool

def run(fn):
    time.sleep(1)
    print(fn * fn)

if __name__ == "__main__":
    test_list = [1, 2, 3, 4, 5, 6]
    # 使用 map 并发执行
    print('使用 map 并行执行:')
    pool = Pool(6)      # 创建拥有 3 个进程数量的进程池
    t1 = time.time()

    # test_list: 要处理的数据列表，run: 处理 test_list 列表中数据的函数
    pool.map(run, test_list)
    pool.close()        # 关闭进程池，不再接受新的进程
    pool.join()         # 主进程阻塞等待子进程的退出

    t2 = time.time()
    print("并行执行时间:%s秒" % int(t2 - t1))

运行结果：

1
2
3
4
5
6
7
8
9
10
使用map并行执行：
14

9
25
3616
并行执行时间：1秒

Process finished with exit code 0

3. 锁

3.1 共享全局变量

多线程共享变量示例：

多进程共享变量示例：

多个进程之间时相互独立的，多个进程之间不共享全局变量，即在一个进程中对全局变量修改过后，不会影响另一个进程中的全局变量。

多进程和多线程最大的区别就在于：

  • 对于多进程，同一个变量各自有一份拷贝存在于每个进程，互不影响。
  • 而多线程不然，所有的线程共用所有的变量。因此，任何一个变量都可以皮任意的一个线程修改。

所以，这就引出了线程资源竞争的问题，这时候锁就派上用场了。

不过在此之前，我们还是先看一个资源竞争的例子帮助我们更好的理解。

1
2
3
4
5
6
7
8
9
10
11
12
13
import threading
import time

a = 0
def add():
    global a
    a += 1
    print('%s adds a to : %d' % (threading.current_thread().name, a))

if __name__ == '__main__':
    for i in range(1, 11):
        t = threading.Thread(name='thread-%d' % (i,), target=add)
        t.start()

运行结果：

1
2
3
4
5
6
7
8
9
10
11
12
thread-1 adds a to:1
thread-2 adds a to:2
thread-3 adds a to:3
thread-4 adds a to:4
thread-5 adds a to:5
thread-6 adds a to:6
thread-7 adds a to:7
thread-8 adds a to:8
thread-9 adds a to:9
thread-10 adds a to:10

Process finished with exit code 0

上面这段代码乍一看似乎没什么问题，但是如果我们稍微变动一下其中的逻辑，模拟一下线程运行过程中一些耗时的情况。此时结果就出现问题了，明显和我们预期的结果不符，线程中的修改操作污染了全局变量的值。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
import threading
import time

a = 0
def add():
    global a
    a += 1
    time.sleep(0.5)
    print('%s adds a to : %d' % (threading.current_thread().name, a))

if __name__ == '__main__':
    for i in range(1, 11):
        t = threading.Thread(name='thread-%d' % (i,), target=add)
        t.start()

运行结果：

1
2
3
4
5
6
7
8
9
10
11
12
thread-2 adds a to:10
thread-8 adds a to:10
thread-3 adds a to:10
thread-7 adds a to:10
thread-1 adds a to:10
thread-4 adds a to:10 thread-6addsa to:10

thread-9 adds a to:10
thread-10 adds a to:10
thread-5 adds a to:10

Process finished with exit code 0

3.2 互斥锁

3.2.1 线程锁

Python 中提供了锁机制，又称互斥锁，而互斥锁为资源引入一个状态：锁定 / 非锁定。

某个线程要更改共享数据时，先将其上锁，此时资源的状态为 “ 锁定 “，其他线程不能更改；直到该线程释放资源，将资源的状态变成 “ 非锁定 “，其他的线程才能再次锁定该资源。

互斥锁保证了每次只有一个线程进行写入操作，即确保在同一时间只有一个线程能够访问共享资源，从而保证了多线程情况下数据的正确性，避免了数据竞争和不一致的问题。

这时候再回到上面的那个例子中，当我们加上锁之后执行，就可以发现结果符合我们的预期。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import threading
import time

lock = threading.Lock()
a = 0
def add():
    global a
    try:
        lock.acquire()      # 获得锁 locked，独占
        a = a + 1
        time.sleep(0.5)     # 延时 0.5 秒，模拟写入所需时间
    finally:
        lock.release()      # 释放锁
    print('%s adds a to : %d' % (threading.current_thread().name, a))

if __name__ == '__main__':
    for i in range(1, 11):
        t = threading.Thread(name='thread-%d' % (i,), target=add)
        t.start()

运行结果：

1
2
3
4
5
6
7
8
9
10
11
12
13
thread-1 adds a to:1
thread-2 adds a to:2
thread-3 adds a to:3
thread-4 adds a to:4
thread-5 adds a to:5
thread-6 adds a to:6
thread-7 adds a to:7
thread-8 adds a to:8
thread-9 adds a to:9
thread-10 adds a to:10

Process finished with exit code 0

3.2.2 进程锁

当多个进程使用同一份数据资源的时候，就会引发数据安全或顺顾序混乱问题。这个时候，我们可以使用 multiprocessing.Lock() 进程锁机制。

进程锁和线程锁的使用方法其实非常类似，只是调用的方法不同。下面这个例子中，多个进程共享一个计数器 counter，通过进程锁 lock 来保证对计数器的修改是互斥的，避免了多个进程同时修改计数器导致的数据不一致问题。

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
import multiprocessing
import time

def worker(lock, counter):
    for _ in range(5):
        # 获取锁
        lock.acquire()
        counter.value += 1
        print(f'进程: {multiprocessing.current_process().name}, 计数器: {counter.value}')
        # 释放锁
        lock.release()
        time.sleep(1)

if __name__ == '__main__':
    # 创建共享的计数器
    counter = multiprocessing.Value('i', 0)
    # 创建进程锁
    lock = multiprocessing.Lock()

    # 创建多个进程
    processes = [
        multiprocessing.Process(target=worker, args=(lock, counter)) for _ in range(3)
    ]

    for p in processes:
        p.start()

    for p in processes:
        p.join()

再来看一个抢票的例子，也可以通过进程锁避免抢票冲突。

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
import multiprocessing
import json
import time

"""
将票数值,存放至 db 文件中,内容为：{"count":1}
"""

def search_ticket():
    dic = json.load(open('db.json'))
    print("查询当前剩余票数为：%s" % dic['count'])

def get_ticket():
    dic = json.load(open('db.json'))
    time.sleep(0.1)  # 模拟读数据延迟
    if dic['count'] > 0:
        dic['count'] -= 1
        time.sleep(0.2)  # 模拟写数据延迟
        json.dump(dic, open('db.json', 'w'))
        print('恭喜你，{}购票成功'.format(multiprocessing.current_process().name))

def task_ticket(lock):
    search_ticket()
    lock.acquire()
    get_ticket()
    lock.release()

if __name__ == '__main__':
    lock = multiprocessing.Lock()
    for i in range(100):  # 模拟并发 100 个客户端抢票
        p = multiprocessing.Process(target=task_ticket, args=(lock,), name='张三-%s' % i)
        p.start()

3.3 死锁

死锁产生的原因：在线程间共享多个资源的时候，当两个或多个线程（或进程）互相等待对方持有的资源，而导致它们都无法继续执行时，就发生了死锁。

3.3.1 死锁示例

我们来看个比较经典的例子，下面的代码就很可能出现死锁的情况：

  • 线程 1 首先获取了 lock1，然后尝试获取 lock2。
  • 同时，线程 2 首先获取了 lock2，然后尝试获取 lock1。
  • 这就可能会导致线程 1 等待线程 2 释放 lock2，线程 2 等待线程 1 释放 lock1，两个线程都无法继续执行，从而形成死锁。
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
import threading

# 创建两个互斥锁
lock1 = threading.Lock()
lock2 = threading.Lock()

def thread1_function():
    lock1.acquire()  # 线程 1 获取锁 1
    print("Thread 1 acquired lock 1")

    # 线程 1 尝试获取锁 2 ，但此时锁 2 可能被线程 2 持有
    lock2.acquire()
    print("Thread 1 acquired lock 2")

    lock1.release()  # 释放锁 1
    lock2.release()  # 释放锁 2

def thread2_function():
    lock2.acquire()  # 线程 2 获取锁 2
    print("Thread 2 acquired lock 2")

    # 线程 2 尝试获取锁 1 ，但此时锁 1 可能被线程 1 持有
    lock1.acquire()
    print("Thread 2 acquired lock 1")

    lock2.release()  # 释放锁 2
    lock1.release()  # 释放锁 1

thread1 = threading.Thread(target=thread1_function)
thread2 = threading.Thread(target=thread2_function)

thread1.start()
thread2.start()

thread1.join()
thread2.join()

运行结果：

1
2
3
4
/venv/bin/python/Users/bytedance/Desktop/project/test_python/test.py
Thread 1 acquired lock1
Thread 2 acquired Lock2

3.3.2 解决死锁

为了避免死锁，可以采用一些策略，比如：

  1. 以固定的顺序获取多个锁。
  2. 尽量减少持有锁的时间，即只在必要的操作期间持有锁。
  3. 使用超时机制来避免无限期地等待锁。

我们这里就通过超时机制来尝试解决一下死锁问题，下述代码中我们为 lock1.acquire() 和 lock2.acquire() 方法都添加了 timeout 参数。这样，如果在指定的时间内无法获取到锁，就会放弃获取并进行相应的处理，从而避免死锁的发生。

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
38
39
40
41
42
43
44
45
import threading
import time

# 创建两个互斥锁
lock1 = threading.Lock()
lock2 = threading.Lock()

def thread1_function():
    lock1.acquire(timeout=5)  # 设置获取锁 1 的超时时间为 5 秒
    print("Thread 1 acquired lock 1")

    try:
        # 尝试在 3 秒内获取锁 2
        if lock2.acquire(timeout=3):
            print("Thread 1 acquired lock 2")
        else:
            # 如果在 3 秒内未获取到锁 2，释放锁 1 并打印错误信息
            lock1.release()
            print("Thread 1 failed to acquire lock 2 within timeout")
    finally:
        lock1.release()  # 确保最终释放锁 1

def thread2_function():
    lock2.acquire(timeout=5)  # 设置获取锁 2 的超时时间为 5 秒
    print("Thread 2 acquired lock 2")

    try:
        # 尝试在 3 秒内获取锁 1
        if lock1.acquire(timeout=3):
            print("Thread 2 acquired lock 1")
        else:
            # 如果在 3 秒内未获取到锁 1，释放锁 2 并打印错误信息
            lock2.release()
            print("Thread 2 failed to acquire lock 1 within timeout")
    finally:
        lock2.release()  # 确保最终释放锁 2

thread1 = threading.Thread(target=thread1_function)
thread2 = threading.Thread(target=thread2_function)

thread1.start()
thread2.start()

thread1.join()
thread2.join()

3.4 GIL 锁

GIL 即全局解释器锁（global interpreter lock），每个线程在执行时候都需要先获取 GIL，保证同一时刻只有一个线程可以执行代码，即同一时刻只有一个线程使用 CPU，也就是说多线程并不是真正意义上的同时执行。

3.4.1 主要特点和影响

  1. 保证线程安全：确保在任意时刻只有一个线程在执行 Python 字节码。这简化了 Python 对共享资源的访问控制，使得内存管理等操作变得更简单，一定程度上保证了线程安全。
  2. 限制多线程并发：由于 GIL 的存在，在多核 CPU 环境下，Python 的多线程不能真正实现并行计算（多个线程同时执行），而只能是并发（交替执行）。对于 CPU 密集型任务（如大量计算），多线程并不能有效提高性能，反而可能因为线程切换的开销而降低性能。
  3. 对 I/O 密集型任务影响较小：对于涉及到网络 I/O 或文件 I/O 的操作，线程在 I/O 等待期间会释放 GIL，允许其他线程运行。所以在 I/O 密集型任务中，多线程仍然可以提高程序的效率。

3.4.2 存在原因

  1. 历史原因：Python 早期的设计决策，为了简化内存管理和线程同步的复杂性。
  2. CPython 实现：CPython（Python 的最常见实现）的内存管理不是线程安全的，GIL 是一种相对简单的解决方案。

3.4.3 应对策略

  1. 对于 CPU 密集型任务，如果想要充分利用多核 CPU，可以使用多进程（multiprocessing 模块），因为每个进程都有自己独立的 GIL。
  2. 对于 I/O 密集型任务，多线程仍然是一个不错的选择。
  3. 也可以考虑使用其他 Python 实现，如 Jython 、IronPython ，它们没有 GIL 的限制，但可能在某些库的支持上不如 CPython 完善。