如何用Python Threading实现3个线程按行交替读取CSV文件

你的代码问题在于每个线程都独立打开并完整读取了整个CSV文件，自然会重复输出所有行。要实现3个线程按行交替读取，需要共享行状态并通过同步机制控制线程执行顺序。

解决方案思路 #

1. 先将CSV所有行加载到内存，避免多线程操作文件指针的同步问题
2. 用共享变量跟踪下一个待读取的行号
3. 使用threading.Condition实现线程间的顺序同步，确保线程按1→2→3→1的循环顺序读取对应行

修改后的代码 #

import threading
import csv

def read_csv(rows, thread_id, lock, condition):
    while True:
        with lock:
            # 等待直到轮到当前线程，或所有行处理完毕
            while (current_row[0] >= len(rows)) or (current_row[0] % 3 != thread_id):
                if current_row[0] >= len(rows):
                    condition.notify_all()
                    return
                condition.wait()
            
            # 读取并打印当前行
            row = rows[current_row[0]]
            print(f"Thread {thread_id + 1} - CSV Row: {row}")
            current_row[0] += 1
            
            # 通知其他线程可以竞争执行
            condition.notify_all()

# 预加载CSV所有行到内存
csv_filename = 'your_csv_file.csv'
with open(csv_filename, 'r') as csv_file:
    rows = list(csv.reader(csv_file))

# 共享状态：当前待处理的行索引（用列表是为了在函数内修改）
current_row = [0]
lock = threading.Lock()
condition = threading.Condition(lock)

# 创建并启动3个线程
threads = []
for i in range(3):
    thread = threading.Thread(target=read_csv, args=(rows, i, lock, condition))
    threads.append(thread)
    thread.start()

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

代码说明 #

• 预加载行：先把CSV内容读到列表rows中，避免多线程同时操作文件带来的同步问题
• 共享行索引：current_row跟踪下一个要读取的行号，用列表是因为Python中不可变类型无法在函数内修改，列表是可变对象可以直接修改
• 条件变量同步：Condition让线程等待直到轮到自己的回合（行号%3等于线程ID），处理完行后通知所有线程，让下一个符合条件的线程继续执行
• 退出逻辑：当行号超过总行数时，线程会退出并通知其他线程结束

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