两个Python进程间实时共享数据的最佳实践及问题求助

嘿，我来帮你搞定这个多进程共享数据的问题！你遇到的本质是进程间资源竞争——当某个进程一直占着共享变量不放时，其他进程根本没机会访问它。下面给你几个实用的解决思路：

1. 用进程安全的队列（Queue）替代直接共享变量 #

这是最推荐的方案，multiprocessing.Queue本身自带锁机制，是天然的进程安全组件。你可以把读取到的数据逐条放进队列，数据库进程和计算进程分别从队列里取数据处理，完全不用担心资源被霸占的问题。

示例代码大概是这样：

from multiprocessing import Process, Queue

def read_data(queue):
    # 模拟从文件读取数据的逻辑
    for line in open("mass_data.txt", "r"):
        data = line.strip()
        queue.put(data)
    queue.put(None)  # 发送任务结束信号

def db_insert(queue):
    while True:
        data = queue.get()
        if data is None:
            queue.put(None)  # 把结束信号传给下一个消费者
            break
        # 执行数据库插入操作
        print(f"插入数据库: {data}")

def data_calculate(queue):
    while True:
        data = queue.get()
        if data is None:
            break
        # 执行数据计算操作
        print(f"计算数据: {data}的哈希值是{hash(data)}")

if __name__ == "__main__":
    data_queue = Queue()
    reader = Process(target=read_data, args=(data_queue,))
    db_worker = Process(target=db_insert, args=(data_queue,))
    calc_worker = Process(target=data_calculate, args=(data_queue,))
    
    reader.start()
    db_worker.start()
    calc_worker.start()
    
    reader.join()
    db_worker.join()
    calc_worker.join()

Queue的put()和get()都是原子操作，两个消费者进程可以各自独立取数据，完全不会互相阻塞。

2. 给共享变量加显式锁（Lock） #

如果你坚持要用Manager()创建的共享变量，那必须手动加锁来控制访问权限。通过multiprocessing.Lock，同一时间只有一个进程能操作共享变量，避免出现霸占的情况。

示例代码：

from multiprocessing import Process, Manager, Lock

def read_data(shared_list, lock):
    for line in open("mass_data.txt", "r"):
        data = line.strip()
        with lock:  # 写入前加锁，操作完自动释放
            shared_list.append(data)
    with lock:
        shared_list.append(None)  # 标记数据读取完成

def db_insert(shared_list, lock):
    while True:
        with lock:
            if not shared_list:
                continue
            data = shared_list.pop(0)
        if data is None:
            with lock:
                shared_list.append(None)
            break
        # 执行数据库插入
        print(f"插入数据库: {data}")

def data_calculate(shared_list, lock):
    while True:
        with lock:
            if not shared_list:
                continue
            data = shared_list.pop(0)
        if data is None:
            break
        # 执行数据计算
        print(f"计算数据: {data}的长度是{len(data)}")

if __name__ == "__main__":
    manager = Manager()
    shared_data = manager.list()
    access_lock = Lock()
    
    reader = Process(target=read_data, args=(shared_data, access_lock))
    db_worker = Process(target=db_insert, args=(shared_data, access_lock))
    calc_worker = Process(target=data_calculate, args=(shared_data, access_lock))
    
    reader.start()
    db_worker.start()
    calc_worker.start()
    
    reader.join()
    db_worker.join()
    calc_worker.join()

注意：锁的范围要尽量小，只在操作共享变量时加锁，别把数据库插入、计算这些耗时操作放到锁里，不然会变成串行执行，浪费多进程的优势。

3. 核心思路总结 #

你之前的问题根源是共享变量没有被正确地做访问控制，导致一个进程长期占用资源。上面两种方案本质都是实现生产者-消费者模式：

• 读取进程是生产者，负责生成数据
• 数据库和计算进程是消费者，负责处理数据
  这种模式能完美解耦数据生产和消费的依赖，从根本上避免资源竞争问题。

内容的提问来源于stack exchange，提问作者weiserhase