Python Threading实现多线程按不同起始位处理JSON用户数据库无效ID

你遇到的重复处理问题本质是没有做多线程场景下的全局任务调度：如果每个线程都单独读取全量用户列表，没有统一的进度追踪机制，所有线程自然都会从第一条数据开始处理，出现大量重复劳动。

采用标准库自带的线程安全队列queue.Queue实现生产者-消费者模式，不需要手动维护进度指针，队列会自动完成任务分配，完全符合你的预期效果。

第一步：核心依赖导入 #

import json
import threading
from queue import Queue

第二步：定义线程处理逻辑 #

# 自定义配置
BATCH_SIZE = 10 # 单批次处理10条数据，对应你举例的10条间隔
VALID_ID_RULE = lambda id: id.startswith("USER_") # 替换为你自己的有效ID校验规则

# 线程安全的结果容器和锁
valid_users = []
write_lock = threading.Lock() # 避免多线程同时写数据导致丢失

def process_task(queue: Queue):
    while True:
        # 从队列取未处理的批次，队列空时自动阻塞等待
        current_batch = queue.get()
        # 收到结束标记则退出线程
        if current_batch is None:
            break

        # 校验当前批次的用户ID
        batch_valid = []
        for user in current_batch:
            if VALID_ID_RULE(user.get("id", "")):
                batch_valid.append(user)
        
        # 加锁写入全局有效用户列表
        with write_lock:
            valid_users.extend(batch_valid)
        
        # 标记当前批次处理完成
        queue.task_done()

第三步：主流程实现 #

if __name__ == "__main__":
    # 1. 加载JSON用户数据库
    with open("user_db.json", "r", encoding="utf-8") as f:
        all_users = json.load(f)
    
    # 2. 按批次拆分用户，存入线程安全队列
    task_queue = Queue()
    for i in range(0, len(all_users), BATCH_SIZE):
        task_queue.put(all_users[i:i+BATCH_SIZE])
    
    # 3. 启动指定数量的工作线程
    THREAD_COUNT = 4 # 可根据你的设备配置调整
    threads = []
    for _ in range(THREAD_COUNT):
        t = threading.Thread(target=process_task, args=(task_queue,))
        t.start()
        threads.append(t)
    
    # 4. 等待所有批次处理完成
    task_queue.join()
    
    # 5. 发送线程退出信号
    for _ in range(THREAD_COUNT):
        task_queue.put(None)
    for t in threads:
        t.join()
    
    # 6. 有效用户写回新的JSON文件
    with open("filtered_user_db.json", "w", encoding="utf-8") as f:
        json.dump(valid_users, f, ensure_ascii=False, indent=2)

• 线程安全队列Queue内部自带锁机制，每次调用get()方法时会自动返回从未被任何线程读取过的任务，完全满足你要的进度自动推进需求：第一个线程取0-9条，第二个取10-19条，第一个处理完后自动取下一个未分配的30-39条，不需要手动维护进度指针。
• 写结果加锁是因为Python原生列表不是线程安全结构，多个线程同时写入会出现数据丢失，加锁后保证同一时间只有一个线程能修改结果列表。
• task_done()和join()配合使用，保证主线程会等所有用户处理完成后才进入写文件步骤，不会提前终止程序。

• 不要在多线程里直接修改原始用户列表，容易出现索引混乱，队列分发的方案容错率更高。
• 如果JSON文件过大无法一次性加载到内存，可以按行读取JSON Lines格式的文件，逐批次放入队列，避免内存占用过高。
• 校验规则可以根据业务需求灵活替换，只需要修改VALID_ID_RULE的逻辑即可。

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