Python如何实现多进程通信？multiprocessing模块详解

multiprocessing.Queue

multiprocessing.Pipe

import multiprocessing
import time
import os

def producer(q, data_count):
    """生产者：生成数据并放入队列"""
    print(f"[{os.getpid()}] 生产者启动...")
    for i in range(data_count):
        item = f"数据块-{i}"
        q.put(item)
        print(f"[{os.getpid()}] 放入: {item}")
        time.sleep(0.1) # 模拟数据生成耗时
    q.put(None) # 发送结束信号
    print(f"[{os.getpid()}] 生产者完成。")

def consumer(q):
    """消费者：从队列中取出数据并处理"""
    print(f"[{os.getpid()}] 消费者启动...")
    while True:
        item = q.get()
        if item is None: # 收到结束信号
            break
        print(f"[{os.getpid()}] 取出: {item}, 正在处理...")
        time.sleep(0.2) # 模拟数据处理耗时
    print(f"[{os.getpid()}] 消费者完成。")

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

    p_process = multiprocessing.Process(target=producer, args=(q, data_to_produce))
    c_process = multiprocessing.Process(target=consumer, args=(q,))

    p_process.start()
    c_process.start()

    p_process.join()
    c_process.join()
    print("所有进程已完成通信示例。")

send()

recv()

import multiprocessing
import time
import os

def sender_process(conn):
    """发送方：通过管道发送数据"""
    print(f"[{os.getpid()}] 发送方启动...")
    for i in range(5):
        msg = f"你好，这是消息 {i}"
        conn.send(msg)
        print(f"[{os.getpid()}] 发送: {msg}")
        time.sleep(0.5)
    conn.send("结束") # 发送结束信号
    conn.close() # 关闭连接
    print(f"[{os.getpid()}] 发送方完成。")

def receiver_process(conn):
    """接收方：通过管道接收数据"""
    print(f"[{os.getpid()}] 接收方启动...")
    while True:
        try:
            msg = conn.recv()
            if msg == "结束":
                break
            print(f"[{os.getpid()}] 接收到: {msg}")
        except EOFError: # 当管道另一端关闭时会抛出
            break
    conn.close()
    print(f"[{os.getpid()}] 接收方完成。")

if __name__ == "__main__":
    parent_conn, child_conn = multiprocessing.Pipe() # 创建管道

    sender = multiprocessing.Process(target=sender_process, args=(parent_conn,))
    receiver = multiprocessing.Process(target=receiver_process, args=(child_conn,))

    sender.start()
    receiver.start()

    sender.join()
    receiver.join()
    print("管道通信示例完成。")

# Manager 示例
import multiprocessing
import time

def worker_with_manager(shared_list, shared_dict, process_id):
    print(f"[{process_id}] 启动...")
    shared_list.append(f"来自进程{process_id}的数据")
    shared_dict[f'key_{process_id}'] = f'value_{process_id}'
    print(f"[{process_id}] 修改了共享数据。")
    time.sleep(0.5)

if __name__ == "__main__":
    with multiprocessing.Manager() as manager:
        shared_list = manager.list() # 创建一个可在进程间共享的列表
        shared_dict = manager.dict() # 创建一个可在进程间共享的字典

        processes = []
        for i in range(3):
            p = multiprocessing.Process(target=worker_with_manager, args=(shared_list, shared_dict, i))
            processes.append(p)
            p.start()

        for p in processes:
            p.join()

        print("\n所有进程完成。")
        print("最终共享列表:", shared_list)
        print("最终共享字典:", shared_dict)

Value

Array

# 共享内存示例 (Value 和 Array)
import multiprocessing
import time

def increment_value(shared_val, process_id):
    print(f"[{process_id}] 启动...")
    for _ in range(5):
        with shared_val.get_lock(): # 使用锁来保护共享值
            shared_val.value += 1
            print(f"[{process_id}] 值增加到: {shared_val.value}")
        time.sleep(0.1)

def modify_array(shared_arr, process_id):
    print(f"[{process_id}] 启动修改数组...")
    for i in range(len(shared_arr)):
        with shared_arr.get_lock():
            shared_arr[i] += process_id
            print(f"[{process_id}] 修改数组[{i}]到: {shared_arr[i]}")
        time.sleep(0.05)

if __name__ == "__main__":
    # 共享整数值
    shared_int = multiprocessing.Value('i', 0) # 'i' 表示有符号整数
    # 共享整数数组
    shared_array = multiprocessing.Array('i', [0, 0, 0]) # 'i' 表示有符号整数，长度为3

    processes = []
    # 针对共享值
    for i in range(2):
        p = multiprocessing.Process(target=increment_value, args=(shared_int, i))
        processes.append(p)
        p.start()

    # 针对共享数组
    for i in range(2):
        p = multiprocessing.Process(target=modify_array, args=(shared_array, i))
        processes.append(p)
        p.start()

    for p in processes:
        p.join()

    print("\n所有进程完成。")
    print("最终共享整数值:", shared_int.value)
    print("最终共享数组:", list(shared_array))

Lock