Python 面试必考题：同步 / 异步、线程 / 进程 / 协程与 GIL ——并发编程全景指南

# 同步代码：顺序执行，耗时≈3s
import time
import requests

def sync_task():
    urls = [f"https://httpbin.org/get?delay=1" for _ in range(3)]
    start = time.time()
    for url in urls:
        requests.get(url)  # 阻塞等待
    print(f"同步耗时：{time.time()-start:.2f}s")  # ≈3s

sync_task()

# 异步代码：并行执行，耗时≈1s
import asyncio
import aiohttp

async def async_task():
    urls = [f"https://httpbin.org/get?delay=1" for _ in range(3)]
    start = time.time()
    async with aiohttp.ClientSession() as session:
        tasks = [session.get(url) for url in urls]
        await asyncio.gather(*tasks)  # 并行发起
    print(f"异步耗时：{time.time()-start:.2f}s")  # ≈1s

asyncio.run(async_task())

# 单线程CPU密集型任务
import time

def cpu_bound(n):
    return sum(i*i for i in range(n))

start = time.time()
result1 = cpu_bound(10**7)
result2 = cpu_bound(10**7)
print(f"单线程耗时：{time.time()-start:.2f}s")  # ≈1.2s

# 多线程CPU密集型任务
import threading

start = time.time()
t1 = threading.Thread(target=cpu_bound, args=(10**7,))
t2 = threading.Thread(target=cpu_bound, args=(10**7,))
t1.start()  # 启动线程
t2.start()
t1.join()   # 阻塞主线程，等待t1完成
t2.join()   # 等待t2完成
print(f"多线程耗时：{time.time()-start:.2f}s")  # ≈1.1s（几乎没有提升，GIL竞争导致）

import threading
import time

def task():
    time.sleep(1)
    print(f"线程{threading.current_thread().name}执行完成")

# 测试1：仅start，不join
print("测试1：仅start，不join")
t1 = threading.Thread(target=task, name="Thread-1")
t1.start()
print("主线程执行完成")  # 主线程会先于子线程结束，子线程仍继续执行

time.sleep(2)  # 等待子线程完成

# 测试2：start+join
print("\n测试2：start+join")
t2 = threading.Thread(target=task, name="Thread-2")
t2.start()
t2.join()  # 主线程等待t2完成
print("主线程执行完成")  # 主线程在子线程完成后执行

# 线程不安全示例：全局变量累加
total = 0

def increment():
    global total
    for _ in range(10**5):
        total += 1  # 非原子操作，会被中断

t1 = threading.Thread(target=increment)
t2 = threading.Thread(target=increment)
t1.start()
t2.start()
t1.join()
t2.join()
print(f"线程不安全的结果：{total}")  # 预期200000，实际≈150000（结果随机）

# 线程安全示例：用Lock
total = 0
lock = threading.Lock()

def increment_safe():
    global total
    with lock:  # 自动获取/释放锁
        for _ in range(10**5):
            total += 1

t1 = threading.Thread(target=increment_safe)
t2 = threading.Thread(target=increment_safe)
t1.start()
t2.start()
t1.join()
t2.join()
print(f"线程安全的结果：{total}")  # 200000（正确）

import multiprocessing
import time

def cpu_bound(n):
    return sum(i*i for i in range(n))

# 多进程CPU密集型任务
start = time.time()
p1 = multiprocessing.Process(target=cpu_bound, args=(10**7,))
p2 = multiprocessing.Process(target=cpu_bound, args=(10**7,))
p1.start()
p2.start()
p1.join()
p2.join()
print(f"多进程耗时：{time.time()-start:.2f}s")  # ≈0.6s（几乎是单线程的1/2，多核并行）

import multiprocessing

def producer(q):
    for item in [1, 2, 3, 4, 5]:
        q.put(item)  # 放入队列
        print(f"生产者放入：{item}")

def consumer(q):
    while True:
        item = q.get()  # 从队列取出，阻塞直到有数据
        if item is None:  # 停止信号
            break
        print(f"消费者取出：{item}")

# 创建队列
q = multiprocessing.Queue()
# 创建进程
p_producer = multiprocessing.Process(target=producer, args=(q,))
p_consumer = multiprocessing.Process(target=consumer, args=(q,))
# 启动进程
p_producer.start()
p_consumer.start()
# 等待生产者完成
p_producer.join()
# 发送停止信号
q.put(None)
# 等待消费者完成
p_consumer.join()

import multiprocessing
import time

def task(n):
    return f"任务{n}完成"

# 创建进程池（默认数量为CPU核心数）
with multiprocessing.Pool() as pool:
    # 批量执行10个任务
    results = pool.map(task, range(10))
    for result in results:
        print(result)

import asyncio
import time

async def coro1():
    await asyncio.sleep(1)
    return "Coro1完成"

async def coro2():
    await asyncio.sleep(2)
    return "Coro2完成"

async def main():
    start = time.time()
    # 并行执行
    result1, result2 = await asyncio.gather(coro1(), coro2())
    print(f"协程1结果：{result1}")
    print(f"协程2结果：{result2}")
    print(f"总耗时：{time.time()-start:.2f}s")  # ≈2s（并行）

asyncio.run(main())