风雪围城

进程、线程和协程


进程，是程序运行时的实例，是操作系统分配资源的基本单位。确保了程序运行的独立性。
线程，是进程中的执行单元，是操作系统调度的最小单位。线程共享进程的内存空间，可以访问进程的资源。
协程，是一种用户态的轻量级”线程”，这种机制的关键在于 Python 中的事件循环。事件循环维护着一个就绪队列，通过不断轮询来检查和调度可以执行的协程。所有的协程都在同一个线程中执行，它们通过主动交出控制权来实现任务切换。

进程和线程

在 Python 中，受限于 GIL( Global Interpreter Lock)，多线程在 CPU 密集型任务上并不能真正实现并行执行。
在任一时刻，只有拥有GIL的线程能执行Python字节码。

这种设计，好处在于：

  • 简化了解释器的实现，便于维护和理解。
  • 避免了线程切换的开销，提高单个线程的执行效率。同时，粗粒度的锁机制，使得 Python 的内存管理模型可以更简单高效。

但是，在 Java 中，线程是可以实现真正意义上的并行执行。这得益于 Java 的线程模型和 JVM 的实现。

  • Java 的线程模型是基于操作系统的原生线程，可以充分利用多核处理器。
  • Java 具备有完善的、细粒度的线程安全机制。比如 Java 的锁机制是基于对象的，可以锁定多线程下的具体代码块。这种锁机制，依托 CAS(Compare And Swap) 原子操作，实现了更加高效的并发控制。比如内存屏障机制，可以保证指令的有序性，防止编译器和处理器对指令进行重排序。

当然，复杂的机制会更好的保证的并发的正确性，但是也会带来性能的下降。

对于 Python 和 Java 来说，它其实体现了两种不同的设计哲学：

  • Python 追求开发效率和代码可读性，强调简单至上。
  • Java 稳定性和工程化优先，强调可维护性与并发。

协程和线程

协程是一种可以被暂停（栈帧）、恢复（栈帧）的函数，由 Python SDK (主要是 asyncio 模块) 实现的语言（软件）层面的特性，在单线程中实现并发，通过事件循来调度任务。

协程函数的定义，通过 async def 关键字定义。遇到 await 关键字时，会暂停执行，将控制权交回事件循环，await 异步操作完成。最终，受事件循环调度，继续执行。

await 关键字后面可以跟一个协程对象，也可以跟一个可等待对象（比如 Future 对象）。

异步转同步

  1. 使用 asyncio.run 运行协程函数

    1
    2
    3
    4
    5
    6
    7
    8
    9
    import asyncio

    async def async_function():
        await asyncio.sleep(1)
        return "异步任务完成"

    # 在同步代码中调用
    result = asyncio.run(async_function())  # 在同步环境运行异步代码
    print(result)

  2. 使用 loop.run_until_complete 运行协程对象
    在已经有事件循环在运行的情况下，使用 loop.run_until_complete 运行协程对象。

    1
    2
    3
    4
    5
    6
    7
    8
    9
    import asyncio

    async def async_function():
        await asyncio.sleep(1)
        return "异步任务完成"

    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(async_function()) 
    print(result)

同步转异步

同步转异步，即将同步函数放到其他线程中执行。
如果有一个阻塞的同步函数，我们可以使用 asyncio.to_thread 将其转换为异步函数。

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

async def async_function():
    await asyncio.sleep(1)
    return "异步任务完成"   

async def main():
    result = await asyncio.to_thread(blocking_function)
    print(result)

asyncio.run(main())

如果同步函数中的计算量非常大（CPU 密集型任务），还可以使用 loop.run_in_executor 让它运行在进程池中。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import asyncio
import time
from concurrent.futures import ProcessPoolExecutor  # 进程池（适用于 CPU 密集型任务）

def cpu_heavy_function():
    """模拟 CPU 密集型同步任务"""
    time.sleep(3)  # 模拟耗时计算
    return "CPU 计算完成"

async def main():
    loop = asyncio.get_running_loop()
    
    # 显式创建进程池
    with ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, cpu_heavy_function)  # 使用进程池
        print(result)

asyncio.run(main())

IO 密集和 CPU 密集

CPU 密集型任务，使用多进程执行。

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

def cpu_bound(n):
    """模拟 CPU 密集型任务"""
    total = 0
    for i in range(n):
        total += i
    return total

def test_cpu():
    start = time.time()
    
    with concurrent.futures.ProcessPoolExecutor() as executor:
        # 使用 executor.map 将 cpu_bound 函数并行执行
        results = list(executor.map(cpu_bound, [10**7] * 4))  # 4 个进程并行执行
    
    print(f"Total CPU tasks took {time.time() - start:.2f} seconds")

test_cpu()

IO 密集型任务，使用异步编程。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import asyncio

async def io_task(n):
    """模拟异步 IO 任务"""
    await asyncio.sleep(2)  # 模拟 IO 操作
    print(f"IO 任务 {n} 完成")

async def test_io():
    start = time.time()
    
    tasks = [io_task(i) for i in range(4)]  # 创建 4 个任务
    await asyncio.gather(*tasks)  # 并发执行

    print(f"Total IO tasks took {time.time() - start:.2f} seconds")

asyncio.run(test_io())

参考链接

Probably the Easiest Tutorial for Python Threads, Processes and GIL

Python Python进程线程协程

© 2025 YueGS