Python多线程实现数据入库求助：多种方案尝试均未生效

你之前的多线程尝试都没找对方向——你是让多个线程重复执行整个批量插入任务，相当于每个线程都把所有数据插一遍，既没提升效率，还可能导致重复数据。真正的优化应该是把单个事件的插入操作拆分到多个线程并行执行，而不是重复跑整个流程。

正确的多线程实现方式（线程池推荐） #

用concurrent.futures.ThreadPoolExecutor来管理线程池，把每个事件的插入作为独立任务提交，这样能并行处理插入操作：

from concurrent.futures import ThreadPoolExecutor
from timeit import default_timer as timer
from api.data.data_endpoint_fetcher import DataEndpointFetcher
from api.models.event import EventModel


class DBAddEvents:
    @classmethod
    def add_single_event(cls, event):
        """单独处理单个事件的插入，提取数据并调用插入方法"""
        try:
            road_name = event['lanelocation']['road']
            avg_speed = event["avgspeed"].get("kmph")
            flow_count = event["flow"].get("count")
            ts_event = event['ts_event']
            uuid = event['measuring_point_id'].get("uuid")
            EventModel.insert_data(road_name, avg_speed, flow_count, ts_event, uuid)
        except Exception as e:
            print(f"处理事件失败: {e}")
            return False
        return True

    @classmethod
    def add_all_events(cls, max_workers=4):
        start = timer()
        combined_events = DataEndpointFetcher.combine_matching_events()
        events_list = combined_events['events']
        
        # 使用线程池并行处理
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            # 把每个事件提交给线程池
            executor.map(cls.add_single_event, events_list)
        
        end = timer()
        print(f"总耗时: {end - start}")


if __name__ == '__main__':
    DBAddEvents.add_all_events(max_workers=4)

为什么你的之前尝试无效？ #

• 第一种尝试：单线程执行整个任务，和原来的单线程代码没区别，只是用Thread包装了一下，完全没并行效果。
• 第二种/第三种尝试：多个线程同时执行add_all_events，每个线程都会加载所有数据并全量插入，导致重复插入数据，而且线程间是重复劳动，反而会因为数据库竞争拖慢速度。

注意事项 #

1. 数据库连接线程安全：确保EventModel.insert_data内部的数据库连接是线程安全的，要么每个线程用独立连接，要么用数据库连接池（比如SQLAlchemy的连接池、psycopg2的连接池等），避免多个线程共享同一个连接导致异常。
2. 异常处理：不要用空的except捕获所有异常，至少要打印异常信息，方便排查问题。
3. 线程数量控制：max_workers不要设置太大，数据库的并发连接数是有限的，过多线程会导致数据库连接阻塞，反而降低效率，一般设置为CPU核心数的2-4倍，或者根据数据库的最大连接数调整。
4. 数据去重：如果你的数据没有唯一约束，并行插入可能会导致重复数据，建议给数据库表的uuid+ts_event这类字段加唯一索引，避免重复插入。

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