并发编程：多线程与多进程

# 并发：多个任务交替执行（看起来像同时）
# 并行：多个任务真正同时执行（多核 CPU）

# 类比：
# - 并发 = 一个人交替做三件事
# - 并行 = 三个人各做一件事

# GIL 确保同一时刻只有一个线程在执行 Python 字节码
# 这意味着：多线程无法利用多核 CPU 加速 CPU 密集型任务

# I/O 密集型任务：线程遇到 I/O 会释放 GIL，因此多线程有效
# CPU 密集型任务：线程一直占用 CPU，无法释放 GIL，因此多线程无效

# 解决方案：
# - I/O 密集型 → 多线程 / asyncio
# - CPU 密集型 → 多进程

import threading
import time

# 方式1：函数方式
def worker(name, delay):
    print(f"线程 {name} 启动")
    time.sleep(delay)
    print(f"线程 {name} 结束")

threads = []
for i in range(3):
    t = threading.Thread(target=worker, args=(f"T{i}", i))
    threads.append(t)
    t.start()

# 等待所有线程结束
for t in threads:
    t.join()

print("所有线程完成")

# 方式2：继承 Thread 类
class WorkerThread(threading.Thread):
    def __init__(self, name, delay):
        super().__init__()
        self.name = name
        self.delay = delay

    def run(self):
        print(f"线程 {self.name} 启动")
        time.sleep(self.delay)
        print(f"线程 {self.name} 结束")

threads = [WorkerThread(f"W{i}", i) for i in range(3)]
for t in threads:
    t.start()
for t in threads:
    t.join()

import threading
import time

def background_task():
    while True:
        print("后台任务运行中...")
        time.sleep(1)

# 守护线程：主线程结束时自动退出
daemon = threading.Thread(target=background_task, daemon=True)
daemon.start()

time.sleep(3)
print("主线程结束")
# 守护线程会自动退出

import threading
import time

def worker():
    time.sleep(1)

t = threading.Thread(target=worker)
print(f"启动前: {t.is_alive()}")  # False
t.start()
print(f"运行中: {t.is_alive()}")  # True
t.join()
print(f"结束后: {t.is_alive()}")  # False

# 当前活跃线程数
print(f"活跃线程: {threading.active_count()}")
# 当前线程
print(f"当前线程: {threading.current_thread().name}")

import threading
import time

# 共享资源
counter = 0
lock = threading.Lock()

def increment():
    global counter
    for _ in range(100000):
        # with 语句自动获取和释放锁
        with lock:
            counter += 1

# 如果没有锁，两个线程同时修改会导致 race condition
threads = [threading.Thread(target=increment) for _ in range(10)]

for t in threads:
    t.start()
for t in threads:
    t.join()

print(f"最终计数: {counter}")  # 1000000

import threading

# Lock：同一个线程重复 acquire 会死锁
# RLock：同一个线程可以多次 acquire

lock = threading.Lock()
rlock = threading.RLock()

def recursive_with_lock(n):
    """使用 Lock 会导致死锁"""
    with lock:
        if n > 0:
            recursive_with_lock(n - 1)  # 第二次 acquire → 死锁！

def recursive_with_rlock(n):
    """使用 RLock 可以重入"""
    with rlock:
        print(f"深度: {n}")
        if n > 0:
            recursive_with_rlock(n - 1)

recursive_with_rlock(3)  # 正常运行

import threading
import time

# 生产者-消费者模式
buffer = []
MAX_SIZE = 5
condition = threading.Condition()

def producer():
    for i in range(10):
        with condition:
            while len(buffer) >= MAX_SIZE:
                print("缓冲区已满，生产者等待...")
                condition.wait()
            buffer.append(f"物品-{i}")
            print(f"生产: 物品-{i}, 缓冲区大小: {len(buffer)}")
            condition.notify()  # 通知消费者
        time.sleep(0.1)

def consumer():
    for _ in range(10):
        with condition:
            while len(buffer) == 0:
                print("缓冲区为空，消费者等待...")
                condition.wait()
            item = buffer.pop(0)
            print(f"消费: {item}, 缓冲区大小: {len(buffer)}")
            condition.notify()  # 通知生产者
        time.sleep(0.2)

prod = threading.Thread(target=producer)
cons = threading.Thread(target=consumer)

prod.start()
cons.start()
prod.join()
cons.join()

import threading
import time

# 控制同时访问资源的线程数量
semaphore = threading.Semaphore(3)  # 最多3个线程同时访问

def limited_access(name):
    with semaphore:
        print(f"{name} 获得访问权限")
        time.sleep(1)
        print(f"{name} 释放访问权限")

threads = [threading.Thread(target=limited_access, args=(f"T{i}",))
           for i in range(6)]

for t in threads:
    t.start()
for t in threads:
    t.join()

import threading
import time

# 用于线程间的一个简单信号
event = threading.Event()

def waiter():
    print("等待事件...")
    event.wait()  # 阻塞直到事件被设置
    print("事件已触发！")

def setter():
    time.sleep(2)
    print("触发事件")
    event.set()

t1 = threading.Thread(target=waiter)
t2 = threading.Thread(target=setter)

t1.start()
t2.start()
t1.join()
t2.join()

import threading
import time
from queue import Queue

def producer(q: Queue, num_items: int):
    for i in range(num_items):
        item = f"数据-{i}"
        q.put(item)
        print(f"生产: {item}")
        time.sleep(0.1)
    q.put(None)  # 发送结束信号

def consumer(q: Queue, name: str):
    while True:
        item = q.get()
        if item is None:  # 收到结束信号
            q.put(None)   # 传递给下一个消费者
            break
        print(f"{name} 消费: {item}")
        time.sleep(0.2)

q = Queue(maxsize=5)
prod = threading.Thread(target=producer, args=(q, 10))
cons1 = threading.Thread(target=consumer, args=(q, "消费者1"))
cons2 = threading.Thread(target=consumer, args=(q, "消费者2"))

prod.start()
cons1.start()
cons2.start()

prod.join()
cons1.join()
cons2.join()

from concurrent.futures import ThreadPoolExecutor, as_completed
import time

def fetch_data(url: str) -> str:
    """模拟网络请求"""
    time.sleep(1)
    return f"数据来自 {url}"

# 创建线程池
with ThreadPoolExecutor(max_workers=5) as executor:
    urls = [f"http://api.example.com/data/{i}" for i in range(10)]

    # 方式1：submit 逐个提交
    futures = [executor.submit(fetch_data, url) for url in urls]

    for future in as_completed(futures):
        result = future.result()
        print(result)

    # 方式2：map 批量提交
    # results = executor.map(fetch_data, urls)
    # for result in results:
    #     print(result)

# 线程池自动关闭
print("所有请求完成")

from concurrent.futures import ThreadPoolExecutor, as_completed
import time
import urllib.request

def download_url(url: str) -> tuple[str, int]:
    """下载 URL 并返回 (url, 字节数)"""
    with urllib.request.urlopen(url, timeout=10) as response:
        data = response.read()
        return url, len(data)

urls = [
    "https://www.python.org",
    "https://www.github.com",
    "https://www.stackoverflow.com",
    "https://pypi.org",
]

start = time.perf_counter()

with ThreadPoolExecutor(max_workers=4) as executor:
    futures = {executor.submit(download_url, url): url for url in urls}
    for future in as_completed(futures):
        url, size = future.result()
        print(f"下载完成: {url} ({size} 字节)")

elapsed = time.perf_counter() - start
print(f"总耗时: {elapsed:.2f}秒")

import multiprocessing
import time

def cpu_intensive(n: int) -> int:
    """CPU 密集型计算"""
    total = 0
    for i in range(n):
        total += i ** 2
    return total

# 方式1：函数方式
if __name__ == "__main__":
    processes = []
    for i in range(4):
        p = multiprocessing.Process(target=cpu_intensive, args=(10_000_000,))
        processes.append(p)
        p.start()

    for p in processes:
        p.join()

    print("所有进程完成")

from concurrent.futures import ProcessPoolExecutor
import time

def compute_square(n: int) -> int:
    """计算平方和"""
    return sum(i ** 2 for i in range(n))

if __name__ == "__main__":
    numbers = [5_000_000, 10_000_000, 15_000_000, 20_000_000]

    start = time.perf_counter()
    with ProcessPoolExecutor(max_workers=4) as executor:
        results = list(executor.map(compute_square, numbers))
    elapsed = time.perf_counter() - start

    for n, result in zip(numbers, results):
        print(f"sum(i^2 for i in range({n})) = {result}")
    print(f"多进程耗时: {elapsed:.2f}秒")

import multiprocessing
import time

def worker(n: int) -> int:
    return n * n

if __name__ == "__main__":
    with multiprocessing.Pool(processes=4) as pool:
        # map：批量执行
        results = pool.map(worker, range(10))
        print(f"map 结果: {results}")

        # apply：单个执行
        result = pool.apply(worker, args=(42,))
        print(f"apply 结果: {result}")

        # starmap：多个参数
        results = pool.starmap(pow, [(2, 10), (3, 5), (4, 3)])
        print(f"starmap 结果: {results}")

        # imap：惰性求值
        for result in pool.imap(worker, range(5)):
            print(f"imap: {result}")

import multiprocessing
import time

def producer(q: multiprocessing.Queue):
    for i in range(5):
        q.put(f"消息-{i}")
        time.sleep(0.1)
    q.put(None)  # 结束信号

def consumer(q: multiprocessing.Queue, name: str):
    while True:
        item = q.get()
        if item is None:
            q.put(None)  # 传递结束信号
            break
        print(f"{name} 收到: {item}")

if __name__ == "__main__":
    q = multiprocessing.Queue()

    p1 = multiprocessing.Process(target=producer, args=(q,))
    p2 = multiprocessing.Process(target=consumer, args=(q, "消费者A"))

    p1.start()
    p2.start()

    p1.join()
    p2.join()

import multiprocessing

def sender(conn: multiprocessing.Connection):
    conn.send("你好")
    conn.send([1, 2, 3])
    conn.close()

def receiver(conn: multiprocessing.Connection):
    while True:
        try:
            msg = conn.recv()
            print(f"收到: {msg}")
        except EOFError:
            break

if __name__ == "__main__":
    parent_conn, child_conn = multiprocessing.Pipe()

    p1 = multiprocessing.Process(target=sender, args=(parent_conn,))
    p2 = multiprocessing.Process(target=receiver, args=(child_conn,))

    p1.start()
    p2.start()

    p1.join()
    p2.join()

import multiprocessing

def worker_counter(counter: multiprocessing.Value, lock: multiprocessing.Lock):
    for _ in range(10000):
        with lock:
            counter.value += 1

def worker_dict(shared_dict: multiprocessing.Manager().dict, key: str):
    shared_dict[key] = key * 10

if __name__ == "__main__":
    # Value：共享数值
    counter = multiprocessing.Value("i", 0)  # 'i' = 有符号整数
    lock = multiprocessing.Lock()

    processes = [
        multiprocessing.Process(target=worker_counter, args=(counter, lock))
        for _ in range(4)
    ]

    for p in processes:
        p.start()
    for p in processes:
        p.join()

    print(f"计数结果: {counter.value}")  # 40000

    # Manager：共享复杂数据结构
    with multiprocessing.Manager() as manager:
        shared_dict = manager.dict()
        processes = [
            multiprocessing.Process(target=worker_dict, args=(shared_dict, f"key-{i}"))
            for i in range(5)
        ]

        for p in processes:
            p.start()
        for p in processes:
            p.join()

        print(f"共享字典: {dict(shared_dict)}")

from concurrent.futures import ThreadPoolExecutor, as_completed
import time

def download_file(url: str) -> tuple[str, str]:
    """模拟文件下载"""
    time.sleep(0.5)  # 模拟网络延迟
    content = f"这是 {url} 的内容"
    return url, content

def parallel_download(urls: list[str], max_workers: int = 5) -> dict[str, str]:
    """并行下载多个文件"""
    results = {}
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(download_file, url): url for url in urls}
        for future in as_completed(futures):
            url = futures[future]
            try:
                _, content = future.result()
                results[url] = content
                print(f"✓ 完成: {url}")
            except Exception as e:
                print(f"✗ 失败: {url} - {e}")
    return results

urls = [f"https://example.com/file-{i}.txt" for i in range(10)]
start = time.perf_counter()
results = parallel_download(urls)
print(f"耗时: {time.perf_counter() - start:.2f}秒")

from concurrent.futures import ProcessPoolExecutor
import time
import math

def is_prime(n: int) -> bool:
    """判断素数"""
    if n < 2:
        return False
    if n == 2:
        return True
    if n % 2 == 0:
        return False
    for i in range(3, int(math.sqrt(n)) + 1, 2):
        if n % i == 0:
            return False
    return True

def find_primes_in_range(start: int, end: int) -> list[int]:
    """在指定范围内查找素数"""
    return [n for n in range(start, end) if is_prime(n)]

if __name__ == "__main__":
    # 单进程 vs 多进程
    total_range = 100_000
    num_workers = 4
    chunk_size = total_range // num_workers

    ranges = [(i * chunk_size, (i + 1) * chunk_size) for i in range(num_workers)]

    # 多进程
    start = time.perf_counter()
    with ProcessPoolExecutor(max_workers=num_workers) as executor:
        futures = [executor.submit(find_primes_in_range, s, e) for s, e in ranges]
        results = []
        for future in futures:
            results.extend(future.result())

    elapsed = time.perf_counter() - start
    print(f"多进程: 找到 {len(results)} 个素数, 耗时 {elapsed:.2f}秒")

import threading
import time

# 没有锁的情况（竞态条件）
counter_without_lock = 0

def race_worker():
    global counter_without_lock
    for _ in range(100000):
        # 读取 → 修改 → 写入（非原子操作）
        temp = counter_without_lock
        counter_without_lock = temp + 1

threads = [threading.Thread(target=race_worker) for _ in range(5)]

for t in threads:
    t.start()
for t in threads:
    t.join()

print(f"无锁: 期望 500000, 实际 {counter_without_lock}")
# 输出可能是: 无锁: 期望 500000, 实际 382941（每次不同）

# 有锁的情况
counter_with_lock = 0
lock = threading.Lock()

def safe_worker():
    global counter_with_lock
    for _ in range(100000):
        with lock:
            counter_with_lock += 1

threads = [threading.Thread(target=safe_worker) for _ in range(5)]

for t in threads:
    t.start()
for t in threads:
    t.join()

print(f"有锁: 期望 500000, 实际 {counter_with_lock}")
# 输出: 有锁: 期望 500000, 实际 500000

import time
import threading
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

def io_task():
    """模拟 I/O 密集型任务"""
    time.sleep(1)  # 模拟 I/O 等待
    return "完成"

def cpu_task(n: int) -> int:
    """模拟 CPU 密集型任务"""
    result = 0
    for i in range(n):
        result += i ** 2
    return result

def compare_performance():
    num_tasks = 8

    # I/O 密集型
    print("=== I/O 密集型任务 ===")

    # 串行
    start = time.perf_counter()
    for _ in range(num_tasks):
        io_task()
    serial_io = time.perf_counter() - start

    # 多线程
    start = time.perf_counter()
    with ThreadPoolExecutor(max_workers=num_tasks) as ex:
        list(ex.map(lambda _: io_task(), range(num_tasks)))
    thread_io = time.perf_counter() - start

    print(f"串行: {serial_io:.2f}s")
    print(f"多线程: {thread_io:.2f}s")
    print(f"加速比: {serial_io / thread_io:.1f}x")

    # CPU 密集型
    print("\n=== CPU 密集型任务 ===")
    n = 5_000_000

    # 串行
    start = time.perf_counter()
    for _ in range(num_tasks):
        cpu_task(n)
    serial_cpu = time.perf_counter() - start

    # 多线程（受 GIL 限制）
    start = time.perf_counter()
    with ThreadPoolExecutor(max_workers=num_tasks) as ex:
        list(ex.map(cpu_task, [n] * num_tasks))
    thread_cpu = time.perf_counter() - start

    # 多进程
    start = time.perf_counter()
    with ProcessPoolExecutor(max_workers=num_tasks) as ex:
        list(ex.map(cpu_task, [n] * num_tasks))
    process_cpu = time.perf_counter() - start

    print(f"串行: {serial_cpu:.2f}s")
    print(f"多线程: {thread_cpu:.2f}s")
    print(f"多进程: {process_cpu:.2f}s")
    print(f"多进程加速比: {serial_cpu / process_cpu:.1f}x")

if __name__ == "__main__":
    compare_performance()

# 决策树（伪代码）
def choose_concurrency_model(task_type: str):
    if task_type == "I/O_bound":
        if need_true_parallelism:
            return "asyncio"  # 单线程事件循环
        else:
            return "threading"  # 多线程
    elif task_type == "CPU_bound":
        return "multiprocessing"  # 多进程
    elif task_type == "mixed":
        return "组合使用"

import threading

def bad_worker():
    raise ValueError("线程中出错")

t = threading.Thread(target=bad_worker)
t.start()
t.join()
# 异常不会传播到主线程！线程会静默退出。

# 解决方案：显式捕获并处理
def safe_worker():
    try:
        # 业务逻辑
        raise ValueError("线程中出错")
    except Exception as e:
        print(f"线程出错: {e}")

import threading

lock_a = threading.Lock()
lock_b = threading.Lock()

def worker_1():
    with lock_a:
        print("Worker 1 持有 lock_a")
        with lock_b:
            print("Worker 1 持有 lock_b")

def worker_2():
    with lock_b:
        print("Worker 2 持有 lock_b")
        with lock_a:  # 可能死锁！
            print("Worker 2 持有 lock_a")

# 解决方案：始终以相同顺序获取锁
def worker_2_fixed():
    with lock_a:  # 与 worker_1 顺序一致
        print("Worker 2 持有 lock_a")
        with lock_b:
            print("Worker 2 持有 lock_b")

import multiprocessing

# BAD: 普通变量不共享
counter = 0

def increment():
    global counter
    counter += 1

if __name__ == "__main__":
    processes = [multiprocessing.Process(target=increment) for _ in range(5)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()
    print(counter)  # 输出 0！子进程有自己的 counter

# GOOD: 使用 Value 或 Manager
def increment_safe(counter):
    counter.value += 1