Python 并发编程

最后编辑于2026/06/10 by:希文捏

️ 阅读须知：请先掌握 Python 基础语法后再阅读本篇！

本篇将带领你逐步深入并发编程的世界，并最终动手实现一个简单的进程池（基于生产者-消费者模型），以解决多任务环境下的资源竞争问题。

进程与线程概念

在编写任何并发代码之前，我们需要先理清两个最基础的概念：进程（Process）与线程（Thread）。它们是操作系统进行资源分配和任务调度的基本单位，也是理解 Python 并发编程的基石。

1. 进程（Process）

进程是操作系统分配资源（如内存空间、文件描述符等）的最小单位。你可以把进程想象成一个正在运行的程序实例。例如，当你在电脑上同时打开浏览器和代码编辑器时，操作系统就会为它们分别创建独立的进程。每个进程都拥有独立的内存空间，彼此之间相互隔离，互不干扰。

2. 线程（Thread）

线程是进程内部的一个执行单元，也是操作系统进行 CPU 调度的最小单位。一个进程可以包含一个或多个线程。如果把进程比作一个工厂，那么线程就是工厂里的工人。工人们共享工厂的场地和设备（即共享进程的内存和资源），但各自执行不同的任务。

3. 并发的意义

当使用得当时，并发编程可以极大地提升代码的运行效率，让程序在处理海量任务时游刃有余。

但是要注意资源竞争问题， 不够优质的代码可能会适得其反

threading 模块

了解了基础概念后，让我们直接开始实践！本章将带你学习如何使用 Python 内置的 threading 模块来实现简单的多线程。

threading 是 Python 的标准库，无需额外安装，导入即可使用。

简单多线程演示

导入threading模块

#使用import 引入threading 
import  threading 
#引入time模块制造延时模拟耗时操作 
import time

示例函数

threading是使用函数或对象来创建多线程任务的， 所以在此之前需要声明一个函数或对象

#导入库
import  threading 
import time 

#声明一个函数 
def worker():
    #输出开始标语
    print("Thread starts running....")
    #使用time.sleep() 来模拟耗时操作（延迟五秒）
    time.sleep(5)
    #输出结束标语
    print("Thread ran successfully....")

创建线程对象

threading.Thread的使用语法为

threadObject = threading.Thread(target=函数名, args=(参数元组,))

• target: 线程启动后要执行的函数。
• args: 传递给 target 函数的参数，接收元组类型。如果只有一个参数，需要写成 (参数,) 的形式。

args的用法这里暂不赘述

#导入库
import  time
import  threading

#示例函数
def worker():
    print("Thread starts running....")
    time.sleep(5)
    print("Thread ran successfully....")


#实例化一个线程对象， 名为thread1 , 使用threading.Thread创建
#Thread类的target参数为的函数名（注意不是调用， 没有括号）
thread1 = threading.Thread(target=worker)
#调用thrad1类的start()方法执行
thread1.start()

运行后， 可以看到正常输出

Thread starts running....
#（五秒后）
Thread ran successfully....

现在， 可以尝试多创建几个线程对象

import  time
import  threading

def worker():
    print("Thread starts running....")
    time.sleep(5)
    print("Thread ran successfully....")


#创建多个线程对象
thread1 = threading.Thread(target=worker)
thread2 = threading.Thread(target=worker)
thread3 = threading.Thread(target=worker)
thread4 = threading.Thread(target=worker)
#运行 
thread1.start()
thread2.start()
thread3.start()
thread4.start()

传递参数

Thread类的args参数可以传递函数参数

import  threading


def worker(id:int):
    print(id)



#									args参数使用元组传递
thread = threading.Thread(target=worker,args=(1,))
thread.start()

使用对象执行

使用对象来执行多线程操作会更加简单规范

import  time
import  threading


#声明一个类， 继承于threading.Thread
class TestThread(threading.Thread):
    #声明类的__init__进行初始化操作 
    def __init__(self, id:int):
        #使用super()来调用父类__init__进行完整的初始化操作 
        super().__init__()
        #参数举例
        self.id = id

	# 当.start()被调用时， run方法会被执行
    def run(self):
        print("Thread starts running....")
        time.sleep(5)
        print("Thread ran successfully....")
	
    # 写一个简单示例方法
    def getID(self):
        return  self.id

#使用新类实例化线程对象
thread1 = TestThread(1)
thread2 = TestThread(2)
thread3 = TestThread(3)
#运行
thread1.start()
thread2.start()
thread3.start()
#等待1秒
time.sleep(1)
#获取线程对象的id 
print(thread1.getID())
print(thread2.getID())
print(thread3.getID())

输出

Thread starts running....
Thread starts running....
Thread starts running....
1
2
3
Thread ran successfully....
Thread ran successfully....
Thread ran successfully....

进程已结束，退出代码为 0

管理线程对象

可以使用列表， 字典等来管理线程对象

import  time
import  threading


class TestThread(threading.Thread):
    def __init__(self, id:int):
        super().__init__()
        self.id = id


    def run(self):
        print("Thread starts running....")
        time.sleep(5)
        print("Thread ran successfully....")

    def getID(self):
        return  self.id

#创建一个列表来存放线程对象
ThreadList = [ ]

#创建线程对象并将其添加进列表中
for i in range(1,5):
    thread = TestThread(i)
    ThreadList.append(thread)

#遍历列表内所有线程对象并调用.start() 方法 
for thread in ThreadList:
    thread.start()

#调用线程对象的方法
for thread in ThreadList:
    print(thread.getID())
    
for thread in ThreadList:
    #join可以阻塞主线程（保证主线程不会在子线程之前就结束）， 直到子线程全部执行完毕 
    thread.join()

输出结果

Thread starts running....
Thread starts running....
Thread starts running....
Thread starts running....
Thread starts running....
Thread starts running....
Thread starts running....
Thread starts running....
Thread starts running....
1
2
3
4
5
6
7
8
9
Thread ran successfully....
Thread ran successfully....
Thread ran successfully....
Thread ran successfully....
Thread ran successfully....
Thread ran successfully....Thread ran successfully....
Thread ran successfully....
Thread ran successfully....

欸？ 为什么 输出结果会有Thread ran successfully通知是重叠的呢？

输出的顺序是混乱的，甚至有时会出现内容"挤在一起"的现象。这是因为多个线程共享同一个标准输出（stdout），这正是资源竞争的一种表现形式——多个线程在争夺同一个共享资源（这里是控制台输出）。

避免资源竞争问题

在并发编程中，避免资源竞争是至关重要的一环。如果处理不当，轻则导致数据错乱，重则引发程序崩溃。

让我们通过一个经典的例子来直观感受资源竞争：

python

import  threading

counter  = 0

def worker():
    global  counter
    for i in range(200000):
        counter  = counter  + 1 # 这里使用 counter = counter + 1 而非 +=，是为了更清晰地展示“读取-修改-写入”的非原子过程
        
threads = []
for i in range(10):
    thread = threading.Thread(target=worker)
    thread.start()
    threads.append(thread)

# 等待所有线程执行完毕
for thread in threads:
    thread.join()

print(counter)

在这个例子中，counter 的理想输出结果应为 200000 × 10 = 2000000。

在实际运行中，你可能会发现结果确实经常是 2000000。这是因为 CPython 解释器中的 GIL（全局解释器锁） 在很大程度上限制了线程的并发，从而“碰巧”掩盖了资源竞争的问题。

你可能连续运行 10 次，甚至 1000 次，得到的结果都是正确的。这就像闯红灯：在多数情况下，可能侥幸没有发生事故，但这并不代表这种行为是安全的。

程序在实际生产环境中往往需要运行成千上万次，甚至面临更高的并发压力。只要存在资源竞争，出错的概率就永远不为零。 我们不能把程序的正确性寄托于 GIL 的保护或运气，而必须通过合理的手段从根本上杜绝隐患。

单线程时程序在修改变量时的操作应是

第一步 :读取 
第二步 :修改 
第三步 :写入（重新赋值）

但是到了多线程环境就截然不同了

线程1:读取 
线程2:读取 
线程1:修改 
线程2:修改 
线程1:写入 
线程3:读取 
线程2:写入 
线程3:修改
线程4:读取 
线程3:写入 
....

这样子， 数据就完全乱掉了

那么该如何防止这种情况的发生呢？目前主流的解决手段是同步（Synchronization），而实现同步最核心的工具就是锁（Lock）

锁的作用，就是将 读取-修改-写入 这一系列操作打包成一个原子操作——即不可被中断的整体。当一个线程获取到锁时，其他线程只能等待，必须等该线程释放锁后才能继续。

当一个线程获取到锁时， 其他线程只能等待。 必须要等获取到锁的线程释放后其他线程才能获取锁

# 加锁后，"读取-修改-写入"成为一个不可分割的整体
线程1: 获取锁
线程2: 等待
线程3: 等待

线程1: 读取 → 修改 → 写入  （整个过程不被打断）

线程1: 释放锁

线程2: 获取锁
线程1: 等待
线程3: 等待

线程2: 读取 → 修改 → 写入

....

Python同样提供了十分方便的方式来操作锁

import  threading

counter  = 0

#创建锁对象
lock = threading.Lock()


def worker():
    global  counter
    for i in range(200000):
        lock.acquire() #获取锁
        counter  = counter  + 1
        lock.release() #释放锁



threads = []
for i in range(10):
    thread = threading.Thread(target=worker)
    thread.start()
    threads.append(thread)

# 等待所有线程执行完毕
for thread in threads:
    thread.join()

print(counter)

这样就可以确保只能有一个线程执行 读取-修改-执行 操作

线程之前通信

TODO 累死了喵， 未完待续喵

multiprocessing 模块

TODO 累死了喵， 未完待续喵