多线程文件备份实时同步更新问题及性能验证咨询

我的目标是将大量文件从文件服务器复制到U盘或其他磁盘备份，写了一个基于tkinter的多线程复制程序，但遇到两个问题：

1. 状态更新不实时：文本控件显示文件已复制时，目标磁盘上实际尚未生成该文件
2. 想验证多线程复制是否比单线程逐一复制更快，不知道具体怎么做

原程序代码：

import tkinter as tk
from tkinter import filedialog
import datetime
import os
import shutil
import threading
from concurrent.futures import ThreadPoolExecutor

count = 0  # Counts copied files
Ex_dirs_list = []  # Excluded directories - those directories won't be copied.
task_counter = 0  # Counter for tracking the number of tasks
counter_lock = threading.Lock()  # Lock for updating the counter safely

def copy_file(srcfile, dstfile):
    # function to copy file from source to destination
    os.makedirs(os.path.dirname(dstfile), exist_ok=True)
    shutil.copy2(srcfile, dstfile)

def copy_files_parallel(srcfile, dstfile):
    global count
    global task_counter

    try:
        copy_file(srcfile, dstfile)
        return True
    except Exception as e:
        StatusText.insert(tk.INSERT, f"Error copying {srcfile} to {dstfile}: {e}")
        StatusText.see("end")
        return False

def copy_directory(src_dir, dst_dir):
    global count
    global task_counter
    excluded_dirs_set = set(os.path.basename(x) for x in Ex_dirs_list)

    dirs = '\n'.join(Ex_dirs_list)
    StatusText.insert(tk.INSERT, f"Coping files from {src_dir} to {dst_dir}.\nIgnores following directories:\n{dirs}\n\n")
    StatusText.see("end")

    start = datetime.datetime.now()
    with ThreadPoolExecutor() as executor:
        futures = []
        for root, dirs, files in os.walk(src_dir):
            base_dir = os.path.basename(root)
            if base_dir in excluded_dirs_set:
                # Skip excluded directories
                StatusText.insert(tk.INSERT, f"Skipping directory: {base_dir}\n")
                StatusText.see("end")
                continue

            dstpath = os.path.join(dst_dir, root[len(src_dir)+1:])
            for file in files:
                srcfile = os.path.join(root, file)
                dstfile = os.path.join(dstpath, file)
                if All_files.get() == 1:  # If all files checked, copy all the files regardless of file changes
                    StatusText.insert(tk.INSERT, f"{srcfile} => {dstfile}: All files\n")
                    StatusText.see("end")
                    futures.append(executor.submit(copy_files_parallel, srcfile, dstfile))
                else:
                    if os.path.exists(dstfile):
                        if os.path.getsize(srcfile) != os.path.getsize(dstfile):
                            StatusText.insert(tk.INSERT, f"{srcfile} => {dstfile}: changed\n")
                            status_text.see("end")
                            futures.append(executor.submit(copy_files_parallel, srcfile, dstfile))
                    else:
                        StatusText.insert(tk.INSERT, f"{srcfile} => {dstfile}: added\n")
                        StatusText.see("end")
                        futures.append(executor.submit(copy_files_parallel, srcfile, dstfile))

    finish = datetime.datetime.now()
    StatusText.insert(tk.INSERT, f"\nSummary:\n=======\nCopied {count} files\nFinished in {(finish - start).total_seconds()} seconds\n\n")
    StatusText.see("end")

def WinBackUp():
    t = threading.Thread(target=copy_directory, args=(FromDir.get(), ToDir.get()))
    t.start()

def ChooseDir(dir):
    global Excluded_dirs

    if dir == "src":
        FromDir.set(filedialog.askdirectory())
    if dir == "dst":
        ToDir.set(filedialog.askdirectory())
    if dir == "exclude":
        directory = filedialog.askdirectory()
        if directory in Ex_dirs_list:
            StatusText.insert(tk.INSERT, f"Error! Can't choose {directory} more than once\n")
        else:
            Ex_dirs_list.append(directory)
            ExcludeText.delete('1.0',"end")
            ExcludeText.insert(tk.END, "\n".join(Ex_dirs_list))
            Excluded_dirs = [os.path.basename(x) for x in Ex_dirs_list]
    if dir == "delete":
        if len(Ex_dirs_list)>=1:
            directory = filedialog.askdirectory()
            if directory in Ex_dirs_list:
                Ex_dirs_list.remove(directory)
                ExcludeText.delete('1.0',"end")
                ExcludeText.insert(tk.END, "\n".join(Ex_dirs_list))
                Excluded_dirs = [os.path.basename(x) for x in Ex_dirs_list]
            else:
                StatusText.insert(tk.INSERT, f"Error! directory {directory} is not in list\n")
        else:
            StatusText.insert(tk.INSERT, f"Error! no directory chosen\n")

win = tk.Tk()
win.geometry("910x350")

ToDir = tk.StringVar()
FromDir = tk.StringVar()
All_files = tk.IntVar()
All_files.set(0)  # Default: copy only the changed files

TilteLabel = tk.Label(win, text="Smart Backup", font=("Arial", 14)).pack()
FromButton = tk.Button(win, text="From", command=lambda:ChooseDir("src")).place(x=10, y=50)
FromEntry = tk.Entry(win, textvariable=FromDir, width=10)
FromEntry.place(x=60, y=50, height=25)

ToButton = tk.Button(win, text="To", command=lambda: ChooseDir("dst")).place(x=150, y=50)
ToEntry = tk.Entry(win, textvariable=ToDir, width=10)
ToEntry.place(x=180, y=50, height=25)

ExcludeButton = tk.Button(win, text="Excluded folders / files", command=lambda: ChooseDir("exclude")).place(x=10, y=90)
ExcludeText = tk.Text(win)
ExcludeText.place(x=10, y=120, height=155, width=250)

DelDirButton = tk.Button(win, text="Del dir", command=lambda: ChooseDir("delete")).place(x=200, y=90)

StatusLabel = tk.Label(win, text="Status")
StatusText = tk.Text(win)
scroll_bar = tk.Scrollbar(win)
scroll_bar.pack(side=tk.RIGHT)
StatusText.place(x=290, y=70, height=205, width=600)
StatusLabel.place(x=290, y=45)

IsAllFiles = tk.Checkbutton(win, text="All files", variable=All_files)
IsAllFiles.place(x=100, y=300)
BackupButton = tk.Button(win, text="Backup", command=WinBackUp).place(x=200, y=300)

win.mainloop()

一、验证多线程 vs 单线程复制速度 #

核心思路 #

用完全相同的源目录、目标目录、排除规则，分别运行单线程和多线程版本，记录总耗时，多次测试取平均值（避免磁盘缓存、系统负载等因素干扰）。

具体实现步骤 #

1. 编写单线程复制函数：逻辑和多线程版本一致，只是去掉ThreadPoolExecutor，直接遍历文件执行复制
2. 添加测试按钮：在UI中新增两个按钮，分别触发单线程和多线程备份，各自记录开始/结束时间并输出对比
3. 控制测试变量：测试前清空目标目录（或使用全新路径），避免增量复制影响结果；重复测试3-5次取平均

单线程复制函数示例

def copy_directory_single_thread(src_dir, dst_dir):
    global count
    excluded_dirs_set = set(os.path.basename(x) for x in Ex_dirs_list)
    # 主线程安全更新状态
    win.after(0, update_status, f"单线程复制：从{src_dir}到{dst_dir}\n忽略目录：\n{chr(10).join(Ex_dirs_list)}\n\n")
    
    start = datetime.datetime.now()
    count = 0  # 重置计数
    
    for root, dirs, files in os.walk(src_dir):
        base_dir = os.path.basename(root)
        if base_dir in excluded_dirs_set:
            win.after(0, update_status, f"跳过目录：{base_dir}\n")
            continue
            
        dstpath = os.path.join(dst_dir, root[len(src_dir)+1:])
        for file in files:
            srcfile = os.path.join(root, file)
            dstfile = os.path.join(dstpath, file)
            need_copy = False
            
            if All_files.get() == 1:
                need_copy = True
            else:
                if not os.path.exists(dstfile) or os.path.getsize(srcfile) != os.path.getsize(dstfile):
                    need_copy = True
                    
            if need_copy:
                win.after(0, update_status, f"🔄 正在复制：{srcfile}\n")
                try:
                    copy_file(srcfile, dstfile)
                    count +=1
                    win.after(0, update_status, f"✅ 已完成：{srcfile}\n")
                except Exception as e:
                    win.after(0, update_status, f"❌ 复制失败：{srcfile} - {str(e)}\n")
    
    finish = datetime.datetime.now()
    win.after(0, update_status, f"\n单线程总结：\n=======\n复制{count}个文件\n耗时{(finish - start).total_seconds()}秒\n\n")

添加测试按钮到UI

# 在原BackupButton下方添加
SingleThreadButton = tk.Button(win, text="单线程测试", command=lambda: threading.Thread(target=copy_directory_single_thread, args=(FromDir.get(), ToDir.get())).start()).place(x=300, y=300)
MultiThreadButton = tk.Button(win, text="多线程测试", command=WinBackUp).place(x=400, y=300)

二、实现状态与复制动作实时同步 #

核心问题分析 #

1. 原程序在提交线程任务时就提前更新了状态，但实际复制动作还在后台执行
2. 多线程直接操作tkinter控件会有线程安全问题，可能导致UI卡顿或崩溃

解决方案 #

1. 调整状态更新时机：等到文件复制完成（或失败）后再更新状态
2. 用win.after()安全更新UI：tkinter要求所有UI操作必须在主线程执行，多线程通过after()方法通知主线程更新
3. 修复线程安全问题：用锁保护全局计数变量count

修正后的核心代码

1. 主线程安全的状态更新函数

def update_status(msg):
    """主线程专用的状态更新函数，避免线程安全问题"""
    StatusText.insert(tk.INSERT, msg)
    StatusText.see("end")

2. 修改后的多线程复制函数

def copy_files_parallel(srcfile, dstfile):
    global count
    try:
        # 先执行复制动作
        copy_file(srcfile, dstfile)
        # 复制完成后通知主线程更新状态
        win.after(0, update_status, f"✅ 已完成：{srcfile} => {dstfile}\n")
        # 线程安全更新计数
        with counter_lock:
            count +=1
        return True
    except Exception as e:
        win.after(0, update_status, f"❌ 复制失败：{srcfile} => {dstfile} - {str(e)}\n")
        return False

3. 修改copy_directory函数的任务提交逻辑

去掉提交任务时的提前状态插入，只保留必要的准备信息：

def copy_directory(src_dir, dst_dir):
    global count
    excluded_dirs_set = set(os.path.basename(x) for x in Ex_dirs_list)
    dirs = '\n'.join(Ex_dirs_list)
    win.after(0, update_status, f"多线程复制：从{src_dir}到{dst_dir}\n忽略目录：\n{dirs}\n\n")
    
    start = datetime.datetime.now()
    count = 0  # 重置计数
    
    # 指定线程池大小，避免线程过多导致I/O竞争
    with ThreadPoolExecutor(max_workers=4) as executor:
        futures = []
        for root, dirs, files in os.walk(src_dir):
            base_dir = os.path.basename(root)
            if base_dir in excluded_dirs_set:
                win.after(0, update_status, f"跳过目录：{base_dir}\n")
                continue
                
            dstpath = os.path.join(dst_dir, root[len(src_dir)+1:])
            for file in files:
                srcfile = os.path.join(root, file)
                dstfile = os.path.join(dstpath, file)
                need_copy = False
                
                if All_files.get() == 1:
                    need_copy = True
                else:
                    if not os.path.exists(dstfile) or os.path.getsize(srcfile) != os.path.getsize(dstfile):
                        need_copy = True
                        
                if need_copy:
                    # 仅提交任务，不提前显示状态
                    win.after(0, update_status, f"🔄 正在复制：{srcfile}\n")
                    futures.append(executor.submit(copy_files_parallel, srcfile, dstfile))
        
        # 等待所有任务完成
        for future in futures:
            future.result()
    
    finish = datetime.datetime.now()
    win.after(0, update_status, f"\n多线程总结：\n=======\n复制{count}个文件\n耗时{(finish - start).total_seconds()}秒\n\n")

额外优化点 #

• 修复原代码中status_text.see("end")的拼写错误（应为StatusText）
• 给ThreadPoolExecutor指定最大线程数（建议4-8，根据磁盘I/O性能调整），避免线程过多导致资源竞争
• 添加"正在复制"的状态提示，让用户明确知道当前进度

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