Python脚本数据多任务学习如何共享

本文目录导读：

1. 多线程共享数据（使用 threading）
2. 多进程共享数据（使用 multiprocessing）
3. 异步IO共享数据（使用 asyncio）
4. 深度学习多任务学习中的参数共享
5. 数据并行中的模型参数共享
6. 最佳实践建议

在Python中实现多任务学习时的数据共享,主要有以下几种常见方法，根据你的具体场景（多线程、多进程、异步IO）选择合适的方案：

多线程共享数据（使用 threading）

由于全局解释器锁(GIL)的存在，多线程适合I/O密集型任务。

使用全局变量 + 锁

import threading
import time
shared_data = {}  # 共享数据
lock = threading.Lock()  # 锁
def worker1():
    for i in range(5):
        with lock:
            shared_data[f"key_{i}"] = f"value_{i}_from_1"
        time.sleep(0.1)
def worker2():
    for i in range(5):
        with lock:
            shared_data[f"key_{i+5}"] = f"value_{i}_from_2"
        time.sleep(0.1)
threads = [threading.Thread(target=worker1), threading.Thread(target=worker2)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(shared_data)

使用 queue.Queue（推荐）

import threading
import queue
import time
data_queue = queue.Queue()
def producer():
    for i in range(5):
        data_queue.put(f"data_{i}")
        time.sleep(0.1)
def consumer():
    while True:
        try:
            data = data_queue.get(timeout=1)
            print(f"Processed: {data}")
            data_queue.task_done()
        except queue.Empty:
            break
producer_thread = threading.Thread(target=producer)
consumer_thread = threading.Thread(target=consumer)
producer_thread.start()
consumer_thread.start()
producer_thread.join()
data_queue.join()  # 等待所有数据处理完成

多进程共享数据（使用 multiprocessing）

由于进程间内存隔离,需要通过特殊机制共享。

Manager 方式（简易但较慢）

from multiprocessing import Process, Manager
def worker(shared_dict):
    for i in range(5):
        shared_dict[i] = i ** 2
if __name__ == "__main__":
    manager = Manager()
    shared_dict = manager.dict()
    processes = [Process(target=worker, args=(shared_dict,)) for _ in range(3)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()
    print(shared_dict)

共享内存方式（高性能）

from multiprocessing import Process, Value, Array
def worker(shared_value, shared_array):
    shared_value.value += 1
    for i in range(len(shared_array)):
        shared_array[i] *= 2
if __name__ == "__main__":
    shared_value = Value('i', 0)  # 'i' 表示整数
    shared_array = Array('d', [1.0, 2.0, 3.0])  # 'd' 表示双精度浮点数
    processes = [Process(target=worker, args=(shared_value, shared_array)) 
                 for _ in range(3)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()
    print(f"Value: {shared_value.value}")
    print(f"Array: {list(shared_array)}")

使用 Queue 进行进程间通信

from multiprocessing import Process, Queue
def producer(q):
    for i in range(5):
        q.put(f"msg_{i}")
def consumer(q):
    while True:
        msg = q.get()
        if msg == "STOP":
            break
        print(f"Received: {msg}")
if __name__ == "__main__":
    q = Queue()
    p1 = Process(target=producer, args=(q,))
    p2 = Process(target=consumer, args=(q,))
    p1.start()
    p2.start()
    p1.join()
    q.put("STOP")  # 通知消费者停止
    p2.join()

异步IO共享数据（使用 asyncio）

适合高并发I/O场景，无需锁机制。

共享变量

import asyncio
shared_data = {}  # 异步中可直接共享，无需锁
async def worker1():
    for i in range(5):
        shared_data[f"key_{i}"] = f"value_{i}_from_1"
        await asyncio.sleep(0.1)
async def worker2():
    for i in range(5):
        shared_data[f"key_{i+5}"] = f"value_{i}_from_2"
        await asyncio.sleep(0.1)
async def main():
    await asyncio.gather(worker1(), worker2())
    print(shared_data)
asyncio.run(main())

使用 asyncio.Queue

import asyncio
async def producer(queue):
    for i in range(5):
        await queue.put(f"data_{i}")
        await asyncio.sleep(0.1)
    await queue.put(None)  # 停止信号
async def consumer(queue):
    while True:
        data = await queue.get()
        if data is None:
            break
        print(f"Processed: {data}")
async def main():
    queue = asyncio.Queue()
    await asyncio.gather(producer(queue), consumer(queue))
asyncio.run(main())

深度学习多任务学习中的参数共享

对于深度学习的多任务学习（如PyTorch）：

import torch
import torch.nn as nn
class SharedEncoder(nn.Module):
    def __init__(self):
        super().__init__()
        self.conv1 = nn.Conv2d(3, 64, 3)
        self.conv2 = nn.Conv2d(64, 128, 3)
    def forward(self, x):
        x = self.conv1(x)
        x = self.conv2(x)
        return x
class TaskSpecificHead(nn.Module):
    def __init__(self, num_classes):
        super().__init__()
        self.fc = nn.Linear(128, num_classes)
    def forward(self, x):
        return self.fc(x)
class MultiTaskModel(nn.Module):
    def __init__(self):
        super().__init__()
        self.shared_encoder = SharedEncoder()  # 共享部分
        self.task1_head = TaskSpecificHead(10)  # 任务1特定
        self.task2_head = TaskSpecificHead(5)   # 任务2特定
    def forward(self, x):
        shared_features = self.shared_encoder(x)
        task1_out = self.task1_head(shared_features)
        task2_out = self.task2_head(shared_features)
        return task1_out, task2_out

数据并行中的模型参数共享

import torch
import torch.nn as nn
# DataParallel 自动处理参数同步
model = nn.Linear(10, 5)
model = nn.DataParallel(model, device_ids=[0, 1, 2])
# 或者使用 DistributedDataParallel（推荐）
import torch.distributed as dist
dist.init_process_group("nccl")
model = nn.parallel.DistributedDataParallel(model)

最佳实践建议

1. 数据量小、读多写少 → 使用 threading + 全局变量 + RLock
2. 需要进程隔离 → 使用 multiprocessing + Manager 或 Queue
3. I/O密集型、高并发 → 使用 asyncio + 普通变量或 asyncio.Queue
4. 深度学习模型共享 → 使用 nn.Module 的子模块参数共享或 DistributedDataParallel
5. 大量数据计算共享 → 使用 numpy 共享内存或 torch.Tensor.share_memory_()

选择时考虑：是否需要进程安全、数据大小、访问频率、性能要求等因素。