import multiprocessing as mp

def worker(name):
    print(f"子进程 {name} 运行中, PID={mp.current_process().pid}")

if __name__ == '__main__':
    p = mp.Process(target=worker, args=('A',))
    p.start()   # 启动子进程
    p.join()    # 等待子进程结束
    print(f"主进程结束, PID={mp.current_process().pid}")

import multiprocessing as mp
import time

def daemon_worker():
    while True:
        print("守护进程运行中...")
        time.sleep(1)

if __name__ == '__main__':
    p = mp.Process(target=daemon_worker, daemon=True)
    p.start()
    time.sleep(3)
    print("主进程结束")
    # 主进程退出时，守护进程自动终止，无需 join

import multiprocessing as mp

def producer(queue):
    for i in range(5):
        item = f"产品-{i}"
        queue.put(item)
        print(f"生产: {item}")
    queue.put(None)  # 哨兵值，通知消费者结束

def consumer(queue):
    while True:
        item = queue.get()
        if item is None:
            break
        print(f"消费: {item}")

if __name__ == '__main__':
    queue = mp.Queue()
    p1 = mp.Process(target=producer, args=(queue,))
    p2 = mp.Process(target=consumer, args=(queue,))
    p1.start()
    p2.start()
    p1.join()
    p2.join()

import multiprocessing as mp

def sender(conn):
    conn.send({"type": "greeting", "msg": "你好！"})
    conn.send({"type": "data", "values": [1, 2, 3]})
    conn.send(None)  # 结束信号

def receiver(conn):
    while True:
        data = conn.recv()
        if data is None:
            break
        print(f"收到: {data}")

if __name__ == '__main__':
    parent_conn, child_conn = mp.Pipe()
    p1 = mp.Process(target=sender, args=(child_conn,))
    p2 = mp.Process(target=receiver, args=(parent_conn,))
    p1.start()
    p2.start()
    child_conn.close()  # 父进程关闭子进程端
    parent_conn.close()  # 子进程关闭父进程端
    p1.join()
    p2.join()

import multiprocessing as mp
import ctypes

def increment(counter, lock):
    for _ in range(100000):
        with lock:
            counter.value += 1

if __name__ == '__main__':
    counter = mp.Value(ctypes.c_int, 0)  # 共享整数
    lock = mp.Lock()
    processes = [mp.Process(target=increment, args=(counter, lock)) for _ in range(4)]
    
    for p in processes:
        p.start()
    for p in processes:
        p.join()
    
    print(f"最终计数: {counter.value}")  # 期望: 400000

import multiprocessing as mp

def worker(shared_dict, key, value):
    shared_dict[key] = value

if __name__ == '__main__':
    with mp.Manager() as manager:
        shared_dict = manager.dict()
        processes = [
            mp.Process(target=worker, args=(shared_dict, f"key-{i}", i * 10))
            for i in range(5)
        ]
        for p in processes:
            p.start()
        for p in processes:
            p.join()
        print(dict(shared_dict))

import multiprocessing as mp
import os

def cpu_bound_task(n):
    """CPU 密集型任务：计算平方和"""
    return sum(i * i for i in range(n))

if __name__ == '__main__':
    numbers = [10**6, 2*10**6, 3*10**6, 4*10**6, 5*10**6]
    
    # 创建进程池（默认为 CPU 核心数）
    with mp.Pool(processes=4) as pool:
        # map: 类似内置 map，阻塞等待所有结果
        results = pool.map(cpu_bound_task, numbers)
        print(f"map 结果: {results}")
        
        # imap: 惰性求值，返回迭代器
        for result in pool.imap(cpu_bound_task, numbers):
            print(f"imap 结果: {result}")
        
        # imap_unordered: 不保证顺序，先完成先返回
        for result in pool.imap_unordered(cpu_bound_task, numbers):
            print(f"imap_unordered 结果: {result}")
        
        # apply_async: 异步提交单个任务
        async_result = pool.apply_async(cpu_bound_task, args=(10**6,))
        print(f"async 结果: {async_result.get()}")
        
        # map_async: 异步版本的 map
        async_results = pool.map_async(cpu_bound_task, numbers)
        print(f"map_async 结果: {async_results.get()}")

import multiprocessing as mp
import time

def slow_task(seconds):
    time.sleep(seconds)
    return f"完成（耗时 {seconds}s）"

if __name__ == '__main__':
    with mp.Pool(3) as pool:
        result = pool.apply_async(slow_task, args=(2,))
        
        # 检查是否完成
        print(f"是否就绪: {result.ready()}")
        
        # 等待结果（带超时）
        try:
            value = result.get(timeout=5)
            print(f"结果: {value}")
        except mp.TimeoutError:
            print("超时！")
        
        # 获取（如果有的话）异常
        # result.get() 会重新抛出子进程中的异常

import multiprocessing as mp

def unsafe_increment(shared_val, n):
    """不加锁的竞争条件示例"""
    for _ in range(n):
        shared_val.value += 1  # 非原子操作！

def safe_increment(shared_val, lock, n):
    """加锁后的安全操作"""
    for _ in range(n):
        with lock:
            shared_val.value += 1

if __name__ == '__main__':
    val = mp.Value('i', 0)
    lock = mp.Lock()
    n = 100000
    
    # 不安全版本
    procs = [mp.Process(target=unsafe_increment, args=(val, n)) for _ in range(4)]
    for p in procs: p.start()
    for p in procs: p.join()
    print(f"不安全结果: {val.value}")  # 不确定，通常 < 400000
    
    val.value = 0
    # 安全版本
    procs = [mp.Process(target=safe_increment, args=(val, lock, n)) for _ in range(4)]
    for p in procs: p.start()
    for p in procs: p.join()
    print(f"安全结果: {val.value}")    # 确定 = 400000

import multiprocessing as mp
import time
import math

def is_prime(n):
    """判断素数"""
    if n < 2:
        return False
    if n < 4:
        return True
    if n % 2 == 0 or n % 3 == 0:
        return False
    i = 5
    while i * i <= n:
        if n % i == 0 or n % (i + 2) == 0:
            return False
        i += 6
    return True

def count_primes_in_range(args):
    """计算 [start, end) 范围内的素数个数"""
    start, end = args
    return sum(1 for n in range(start, end) if is_prime(n))

def benchmark():
    total = 5_000_000
    chunk_size = total // mp.cpu_count()
    ranges = [
        (i * chunk_size, min((i + 1) * chunk_size, total))
        for i in range(mp.cpu_count())
    ]
    
    # 单进程
    start = time.perf_counter()
    single_result = count_primes_in_range((0, total))
    single_time = time.perf_counter() - start
    
    # 多进程
    start = time.perf_counter()
    with mp.Pool() as pool:
        results = pool.map(count_primes_in_range, ranges)
    multi_result = sum(results)
    multi_time = time.perf_counter() - start
    
    print(f"CPU 核心数: {mp.cpu_count()}")
    print(f"单进程: {single_result} 个素数, 耗时 {single_time:.2f}s")
    print(f"多进程: {multi_result} 个素数, 耗时 {multi_time:.2f}s")
    print(f"加速比: {single_time / multi_time:.2f}x")

if __name__ == '__main__':
    benchmark()

import multiprocessing as mp
import time
import os

def heavy_computation(image_id):
    """模拟 CPU 密集型图像处理"""
    # 模拟计算密集操作
    result = 0
    for i in range(10_000_000):
        result += (i * image_id) % 997
    return {"image_id": image_id, "result": result, "pid": os.getpid()}

def process_images(image_ids):
    """使用进程池处理图片"""
    with mp.Pool(processes=min(4, mp.cpu_count())) as pool:
        results = pool.map(heavy_computation, image_ids)
    return results

if __name__ == '__main__':
    image_ids = list(range(16))
    
    start = time.perf_counter()
    results = process_images(image_ids)
    elapsed = time.perf_counter() - start
    
    # 统计各进程处理数量
    pid_counts = {}
    for r in results:
        pid = r["pid"]
        pid_counts[pid] = pid_counts.get(pid, 0) + 1
    
    print(f"处理 {len(image_ids)} 张图片，耗时 {elapsed:.2f}s")
    print(f"使用进程数: {len(pid_counts)}")
    for pid, count in pid_counts.items():
        print(f"  PID {pid}: 处理 {count} 张")

# 显式指定启动方式
if __name__ == '__main__':
    mp.set_start_method('spawn')  # 全局只能调用一次
    # 或
    ctx = mp.get_context('spawn')
    p = ctx.Process(target=worker)

# ❌ 错误：lambda 不能被 pickle
pool.map(lambda x: x**2, [1, 2, 3])

# ✅ 正确：使用普通函数
def square(x):
    return x ** 2
pool.map(square, [1, 2, 3])

mp.Process(target=func, args=(...), kwargs={...}, daemon=False)
# 启动: start()
# 等待: join(timeout=None)
# 终止: terminate() / kill()
# 状态: is_alive(), pid, exitcode

mp.Queue(maxsize=0)
# 放入: put(obj, block=True, timeout=None) / put_nowait(obj)
# 取出: get(block=True, timeout=None) / get_nowait()
# 状态: qsize(), empty(), full()

mp.Pool(processes=None)
# 同步: map(func, iterable) / starmap(func, iterable)
# 异步: map_async(func, iterable) / apply_async(func, args)
# 惰性: imap(func, iterable) / imap_unordered(func, iterable)
# 控制: close() / terminate() / join()