Python 中的进程(Process)实例详解

# 进程的独立内存空间
from multiprocessing import Process
data = []  # 全局列表
def worker():
    data.append(1)  # 每个进程有自己 copy 的 data
    print(f"{id(data)}: {data}")
p1 = Process(target=worker)
p2 = Process(target=worker)
p1.start()
p2.start()
p1.join()
p2.join()
# 输出：不同内存地址的进程各自操作自己的 data

from multiprocessing import Process
import os
# 方式1：传入目标函数
def worker(name, num):
    print(f"进程 {name}: PID={os.getpid()}, 父进程={os.getppid()}")
    return num * 2
p = Process(target=worker, args=("worker1", 100))
p.start()      # 启动进程
p.join()       # 等待进程结束
print(f"exit code: {p.exitcode}")
# 方式2：自定义进程类
class MyProcess(Process):
    def __init__(self, name):
        super().__init__()
        self.name = name
    def run(self):
        print(f"运行进程: {self.name}")
p = MyProcess("my_worker")
p.start()
p.join()

import multiprocessing as mp
# 设置启动方式（必须在 if __name__ == '__main__' 内）
if __name__ == '__main__':
    mp.set_start_method('spawn')  # Windows 默认，macOS 可选
    # 或 mp.set_start_method('fork')   # Linux 默认
# 三种启动方式对比：
# | 方式    | 特点                           | 适用平台    |
# |--------|-------------------------------|-----------|
# | spawn  | 全新启动，不继承父进程资源       | Windows    |
# | fork   | 复制父进程，继承资源（不安全）    | Linux/macOS|
# | forkserver| 通过服务进程 fork（安全）       | Unix-like  |

from multiprocessing import Pool
import time
def cpu_intensive(n):
    """CPU 密集型任务"""
    return sum(i * i for i in range(n))
def io_bound(n):
    """I/O 密集型任务"""
    time.sleep(n)
    return n
if __name__ == '__main__':
    # 创建进程池（默认 CPU 核心数）
    with Pool(processes=4) as pool:
        # 方式1：map（阻塞，保持顺序）
        results = pool.map(cpu_intensive, [10**7] * 4)
        # 方式2：map_async（非阻塞）
        result = pool.map_async(cpu_intensive, [10**7] * 4)
        print("异步提交完成")
        results = result.get()  # 等待结果
        # 方式3：apply（同步）
        result = pool.apply(cpu_intensive, (10**7,))
        # 方式4：apply_async（异步）
        async_result = pool.apply_async(cpu_intensive, (10**7,))
        result = async_result.get(timeout=30)
        # 方式5：starmap（多参数）
        args = [(1, 2), (3, 4), (5, 6)]
        results = pool.starmap(lambda a, b: a + b, args)

from multiprocessing import Process, Queue
import time
def producer(q):
    for i in range(5):
        q.put(f"消息{i}")
        print(f"生产: {i}")
        time.sleep(0.5)
    q.put(None)  # 结束标志
def consumer(q):
    while True:
        msg = q.get()
        if msg is None:
            break
        print(f"消费: {msg}")
        time.sleep(1)
if __name__ == '__main__':
    q = Queue(maxsize=10)  # 指定最大容量
    p1 = Process(target=producer, args=(q,))
    p2 = Process(target=consumer, args=(q,))
    p1.start()
    p2.start()
    p1.join()
    p2.join()

from multiprocessing import Process, Pipe
def worker(conn):
    """子进程通过管道发送数据"""
    conn.send([1, 2, 3])      # 发送对象
    print(f"收到父进程消息: {conn.recv()}")
    conn.close()
if __name__ == '__main__':
    # 创建双向管道
    parent_conn, child_conn = Pipe()
    p = Process(target=worker, args=(child_conn,))
    p.start()
    print(f"收到子进程数据: {parent_conn.recv()}")
    parent_conn.send("来自父进程的问候")
    p.join()

from multiprocessing import Process, Value, Array
def worker(val, arr):
    val.value += 1
    for i in range(len(arr)):
        arr[i] += i
if __name__ == '__main__':
    # 创建共享变量
    num = Value('i', 0)      # 'i' 表示有符号整数
    arr = Array('d', [0.0, 1.0, 2.0])  # 'd' 表示 double
    processes = []
    for _ in range(4):
        p = Process(target=worker, args=(num, arr))
        processes.append(p)
        p.start()
    for p in processes:
        p.join()
    print(f"num: {num.value}")
    print(f"arr: {arr[:]}")

from multiprocessing import Process, Manager
def worker(d, l, key, value):
    d[key] = value
    l.append(value)
if __name__ == '__main__':
    with Manager() as manager:
        # 创建共享字典和列表
        d = manager.dict()
        l = manager.list()
        processes = []
        for i in range(5):
            p = Process(target=worker, args=(d, l, f"key{i}", i))
            processes.append(p)
            p.start()
        for p in processes:
            p.join()
        print(f"dict: {d}")
        print(f"list: {l}")

from multiprocessing import Process, Lock, Value
import time
def worker(lock, counter):
    for _ in range(1000):
        lock.acquire()
        try:
            counter.value += 1
        finally:
            lock.release()
if __name__ == '__main__':
    counter = Value('i', 0)
    lock = Lock()
    processes = [Process(target=worker, args=(lock, counter)) for _ in range(10)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()
    print(f"最终值: {counter.value}")  # 应该是 10000