为什么需要关注线程安全

在Python多线程编程中传递列表时，必须特别关注线程安全问题。由于Python的全局解释器锁(GIL)，多个线程不能同时执行Python字节码，但列表操作不是原子操作。这意味着多个线程同时修改同一个列表可能导致：

• 数据不一致或损坏
• 丢失更新（一个线程的修改覆盖另一个线程的修改）
• 程序崩溃或不可预测的行为

因此，在传递列表到线程时，我们需要使用适当的同步机制来确保线程安全。

方法1：直接传递列表（只读场景）

如果列表只需要被线程读取而不修改，可以直接将列表作为参数传递给线程函数。这是最简单高效的方法。

import threading

def process_data(data_list, results, index):
    """线程处理函数：计算平方"""
    # 只读取列表，不修改
    num = data_list[index]
    results[index] = num * num

# 创建共享数据结构
original_data = [1, 2, 3, 4, 5]
results = [0] * len(original_data)  # 预初始化结果列表

# 创建并启动线程
threads = []
for i in range(len(original_data)):
    t = threading.Thread(target=process_data, args=(original_data, results, i))
    t.start()
    threads.append(t)

# 等待所有线程完成
for t in threads:
    t.join()

print("原始数据:", original_data)
print("计算结果:", results)  # 输出: [1, 4, 9, 16, 25]

方法2：使用Queue队列（生产者-消费者模式）

Queue是线程安全的先进先出(FIFO)数据结构，非常适合在多线程间安全传递数据。

import threading
import queue
import time

def producer(q, items):
    """生产者线程：将数据放入队列"""
    for item in items:
        print(f"生产者添加: {item}")
        q.put(item)
        time.sleep(0.1)  # 模拟生产耗时

def consumer(q, name):
    """消费者线程：从队列获取并处理数据"""
    while True:
        item = q.get()
        if item is None:  # 终止信号
            break
        print(f"{name} 处理: {item * 2}")
        time.sleep(0.2)  # 模拟处理耗时
        q.task_done()  # 标记任务完成

# 创建线程安全队列
q = queue.Queue()

# 创建生产者线程
prod_thread = threading.Thread(target=producer, args=(q, [1, 2, 3, 4, 5]))
prod_thread.start()

# 创建消费者线程
consumer_threads = []
for i in range(2):  # 2个消费者
    t = threading.Thread(target=consumer, args=(q, f"消费者-{i+1}"))
    t.start()
    consumer_threads.append(t)

# 等待生产者完成
prod_thread.join()

# 等待队列清空
q.join()

# 停止消费者线程
for _ in range(len(consumer_threads)):
    q.put(None)  # 发送终止信号

for t in consumer_threads:
    t.join()

方法3：使用线程锁(Lock)保护共享列表

当多个线程需要修改同一个列表时，使用Lock可以确保每次只有一个线程访问共享资源。

import threading
import random

# 共享数据
shared_list = []
list_lock = threading.Lock()

def add_to_list(item):
    """线程安全地向列表添加元素"""
    with list_lock:  # 自动获取和释放锁
        # 临界区 - 每次只有一个线程执行此代码块
        shared_list.append(item)
        print(f"添加 {item}, 列表: {shared_list}")

def process_items(items):
    """处理一组项目并添加到共享列表"""
    for item in items:
        # 模拟一些处理工作
        processed = item * 10 + random.randint(1, 5)
        add_to_list(processed)

# 创建线程
threads = []
data_sets = [
    [1, 2, 3],
    [4, 5, 6],
    [7, 8, 9]
]

for data in data_sets:
    t = threading.Thread(target=process_items, args=(data,))
    t.start()
    threads.append(t)

# 等待所有线程完成
for t in threads:
    t.join()

print("最终共享列表:", sorted(shared_list))

方法4：使用Manager对象（进程间共享）

当使用多进程而非多线程时，Manager对象可以创建在进程间共享的数据结构。

import multiprocessing

def worker(shared_list, start, end):
    """工作进程：计算平方并添加到共享列表"""
    for i in range(start, end):
        result = i * i
        shared_list.append(result)

if __name__ == '__main__':
    # 创建Manager和共享列表
    with multiprocessing.Manager() as manager:
        shared_list = manager.list()  # 进程间共享的列表
        
        processes = []
        chunk_size = 5
        total_items = 20
        
        # 创建并启动进程
        for i in range(0, total_items, chunk_size):
            p = multiprocessing.Process(
                target=worker,
                args=(shared_list, i, min(i + chunk_size, total_items))
            p.start()
            processes.append(p)
        
        # 等待所有进程完成
        for p in processes:
            p.join()
        
        # 转换为普通列表并排序
        final_list = sorted(shared_list)
        print("计算结果:", final_list)

性能比较与选择建议

常见问题解答