如何随机打乱csv.DictReader的输出并结合ThreadPoolExecutor实现多线程处理不同行数据

我来帮你解决这个问题～你现在遇到的核心问题是每个线程都重复读取了原始顺序的CSV行，所以输出和处理逻辑完全一致。下面给你两种可行的解决方案，分别对应不同的需求场景：

方案一：每个线程独立读取并打乱全行（适合压力测试等重复处理场景） #

如果你的需求是让每个线程都遍历所有行，但顺序随机（比如做接口压力测试），可以在每个线程里把读取到的行转成列表后打乱，再进行处理。这样每个线程的行顺序都是随机且不同的：

import csv
import random
from concurrent.futures import ThreadPoolExecutor
import requests

def grab_data(thread_id):
    with open('devices.csv', 'r') as csv_file:
        csv_reader = csv.DictReader(csv_file)
        # 把CSV行转成列表，原地打乱顺序
        rows = list(csv_reader)
        random.shuffle(rows)
        
        # 遍历打乱后的行发送请求
        for line in rows:
            r = requests.get(
                url=f"https://192.168.1.56:1813/rest/{line['A']}/sh/1/sl/{line['B']}/{line['C']}/status?",
                verify=False,
                auth=('random', 'TestTest'),
            )
            # 可选：打印线程ID和处理结果，方便调试
            print(f"线程{thread_id}处理行: {line}，响应状态码: {r.status_code}")

with ThreadPoolExecutor(max_workers=5, thread_name_prefix="API") as executor:
    executor.map(grab_data, range(1, 15))

方案二：一次性读取并打乱，分发给线程处理不同子集（适合并行处理无重复场景） #

如果你想要所有线程同时处理不同的行（避免重复处理同一行），更高效的方式是先一次性读取所有行并打乱，再分割成多个子集分配给每个线程。这样既避免了重复读取文件，又能保证每个线程处理的是不同的随机行：

import csv
import random
from concurrent.futures import ThreadPoolExecutor
import requests

def process_rows(thread_id, rows_subset):
    # 处理分配到的行子集
    for line in rows_subset:
        r = requests.get(
            url=f"https://192.168.1.56:1813/rest/{line['A']}/sh/1/sl/{line['B']}/{line['C']}/status?",
            verify=False,
            auth=('random', 'TestTest'),
        )
        print(f"线程{thread_id}处理行: {line}，响应状态码: {r.status_code}")

if __name__ == "__main__":
    # 提前读取所有行并打乱，只做一次更高效
    with open('devices.csv', 'r') as csv_file:
        csv_reader = csv.DictReader(csv_file)
        all_rows = list(csv_reader)
        random.shuffle(all_rows)
    
    max_workers = 5
    # 把打乱后的行分割成对应线程数的子集
    chunk_size = len(all_rows) // max_workers
    row_chunks = []
    for i in range(max_workers):
        start_idx = i * chunk_size
        # 最后一个子集包含剩余所有行，避免遗漏
        end_idx = start_idx + chunk_size if i != max_workers -1 else len(all_rows)
        row_chunks.append(all_rows[start_idx:end_idx])
    
    # 启动线程池分配任务
    with ThreadPoolExecutor(max_workers=max_workers, thread_name_prefix="API") as executor:
        executor.map(process_rows, range(1, max_workers+1), row_chunks)

一些注意事项 #

• 记得导入random模块，random.shuffle()会原地打乱列表，不需要额外赋值。
• 如果你觉得verify=False的不安全警告很烦，可以在代码开头加上这一行关闭警告：

  requests.packages.urllib3.disable_warnings(requests.packages.urllib3.exceptions.InsecureRequestWarning)
  

• 如果你的CSV文件特别大，第二种方案更节省资源（只读取一次文件），但两种方案都需要将全行加载到内存；如果文件超大到内存放不下，可以考虑分块读取并打乱，但一般业务场景下全行加载是没问题的。

备注：内容来源于stack exchange，提问作者UndyingThanos