Python多线程写入CSV文件表头重复、内容覆盖问题如何解决

你遇到的两个写入异常来自两个逻辑错误，另外现有写法还藏着多线程写文件的并发风险：

• 表头写入逻辑被放在了每条数据都会调用的saving_statistics函数内，使用a追加模式时，每写一条检测结果就会重复插入一次表头
• 使用w覆盖模式时，每次调用saving_statistics都会重新打开文件清空已有内容，最终只会保留最后一次写入的内容
• 多线程同时操作同一个文件句柄，没有做互斥控制的话，高并发场景下会出现内容交错、行错位、写入丢失的问题

方案1：边检测边写入（适合需要实时落盘结果的场景） #

调整逻辑，把表头写入移到线程池启动前执行，新增线程锁保证写入互斥，同时移除saving_statistics内的表头写入逻辑：

import csv
import concurrent.futures
import threading

# 初始化线程写锁，避免并发写入冲突
write_lock = threading.Lock()

def saving_statistics(save_result, saving):
    with write_lock:
        # 追加模式只写结果行，不写表头
        with open(saving, 'a', encoding='utf-8', newline='') as csv_file:
            csv_writer = csv.writer(csv_file, delimiter=';')
            csv_writer.writerow(save_result)

# 你的业务逻辑入口
if __name__ == "__main__":
    # 替换为你实际的文件路径、IP列表变量
    saving = "connection_check_result.csv"
    # ipa_str_s = 你原有的IP地址列表

    # 线程启动前先写表头，w模式自动清空旧文件的错误内容
    with open(saving, 'w', encoding='utf-8', newline='') as csv_file:
        csv_writer = csv.writer(csv_file, delimiter=';')
        csv_writer.writerow(['IP-address', 'Packets_transmitted', 'Packets_received'])

    # 原线程池逻辑不需要改动
    with concurrent.futures.ThreadPoolExecutor() as executor:
        futures = []
        for addresses in ipa_str_s:
            futures.append(executor.submit(checking_connection, ip_address=addresses))
        for future in concurrent.futures.as_completed(futures):
            save_result = future.result()
            saving_statistics(save_result, saving)

注：打开CSV文件时添加newline=''是Python csv模块官方推荐写法，可以避免Windows系统下生成的文件出现多余空行。

方案2：全量收集后一次性写入（逻辑更简单，无并发风险，优先推荐） #

如果不需要边检测边落盘，可以等所有检测任务执行完成后，在主线程一次性把所有结果写入文件，完全不需要处理锁逻辑：

import csv
import concurrent.futures

# 你的业务逻辑入口
if __name__ == "__main__":
    saving = "connection_check_result.csv"
    # ipa_str_s = 你原有的IP地址列表

    all_results = []
    with concurrent.futures.ThreadPoolExecutor() as executor:
        futures = []
        for addresses in ipa_str_s:
            futures.append(executor.submit(checking_connection, ip_address=addresses))
        # 收集所有检测结果
        for future in concurrent.futures.as_completed(futures):
            all_results.append(future.result())

    # 主线程一次性写入表头+所有结果
    with open(saving, 'w', encoding='utf-8', newline='') as csv_file:
        csv_writer = csv.writer(csv_file, delimiter=';')
        csv_writer.writerow(['IP-address', 'Packets_transmitted', 'Packets_received'])
        csv_writer.writerows(all_results)

两种方案最终生成的文件都完全符合你预期的格式，不会出现重复表头、内容被覆盖的问题。

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