多线程运行结果重复而非顺序输出，代码问题排查求助

问题描述 #

在Python多线程编程中，输入线程数量后启动多个线程执行read_email_from_gmail函数：

• 测试代码期望输出one到ten的顺序结果，但实际结果重复且无序
• 附上测试代码及Gmail邮件读取的原始代码，确认代码是否存在问题

测试代码 #

def read_email_from_gmail():
    data = ["one","two","three",'four','five','six','seven','eight','nine','ten']
    for x in data:
        print(f'{x} \n')

threads = input("\n Amount of Thread ?: ")
for i in range(int(threads)):
    threading.Thread(target=read_email_from_gmail).start()

原始代码 #

def read_email_from_gmail():
    try:
        mail = imaplib.IMAP4_SSL(SMTP_SERVER)
        mail.login(ORG_EMAIL,FROM_PWD)
        mail.select('inbox')
    
        data = mail.search(None, '(Subject "verify email")' )
        mail_ids = data[1]
        id_list = mail_ids[0].split()
        if not id_list:
            print("No Email Found")
        else: 
            print(f"{Fore.LIGHTWHITE_EX}----------------------------------------{Fore.RESET}")
            print(f"{Fore.LIGHTMAGENTA_EX}            Found {len(id_list)} Email{Fore.RESET}")
            print(f"{Fore.LIGHTWHITE_EX}----------------------------------------{Fore.RESET}")
            first_email_id = int(id_list[0])
            latest_email_id = int(id_list[-1])
            x = latest_email_id + 1
            n = 0
            for i in range(first_email_id,x):
                data = mail.fetch(str(i), '(RFC822)' )
                for response_part in data:
                    arr = response_part[0]
                    if isinstance(arr, tuple):
                        msg = email.message_from_string(str(arr[1],'utf-8'))
                        for part in msg.walk():
                            if part.get_content_type() == 'text/plain':
                                plain_text = part.get_payload(decode=True)
                                str_enc = str(plain_text)
                                extractor = URLExtract()
                                extractor.update()
                                urls = extractor.find_urls(str_enc)
                                str_url = str(urls).replace("\\","").replace("rnrn'","").replace('"','').replace("[","").replace("]","").replace("'","")
                                to = msg['to']
                                rancolor = random.choice(color)
                                print(f"[{Fore.MAGENTA}{n}{Fore.RESET}] {rancolor}[ {to} ] Found Verification Link {Fore.RESET}")
                                n += 1
                                save = open("list.txt", "a")
                                save.write(f"{str_url}\n")
                                save.close()

问题分析 #

测试代码的核心问题 #

1. 重复执行逻辑：每个线程都会完整运行read_email_from_gmail函数，开N个线程就会重复打印N遍one到ten，这是当前代码的必然结果
2. 无序输出原因：多线程的执行顺序由操作系统调度决定，线程间的打印操作会被互相打断穿插，导致输出顺序混乱

原始代码的潜在问题 #

1. 重复处理邮件：多个线程会独立建立Gmail连接、搜索相同主题的邮件，最终重复提取同一批邮件的链接到list.txt
2. 文件写入竞争：多个线程同时写入list.txt，没有线程安全保护，会导致文件内容错乱、部分内容丢失
3. 资源未释放：IMAP连接未在线程结束时关闭，长期运行会耗尽Gmail的连接配额
4. 低效初始化：每次处理邮件都重新创建URLExtract实例，造成不必要的性能损耗

解决建议 #

针对测试代码（实现顺序无重复输出） #

如果需要多线程协作输出one到ten而非重复打印，可将数据拆分给不同线程处理：

import threading

def print_data(start, end, data):
    for x in data[start:end]:
        print(f'{x}\n')

data = ["one","two","three",'four','five','six','seven','eight','nine','ten']
threads_num = int(input("\n Amount of Thread ?: "))
step = len(data) // threads_num
threads = []

for i in range(threads_num):
    start = i * step
    end = start + step if i != threads_num-1 else len(data)
    t = threading.Thread(target=print_data, args=(start, end, data))
    threads.append(t)
    t.start()

# 等待所有线程执行完毕
for t in threads:
    t.join()

针对原始代码（避免重复与资源竞争） #

1. 拆分任务流程：单线程获取所有邮件ID，再将ID分片分配给多线程处理，避免重复搜索邮件
2. 文件写入加锁：用线程锁确保同一时间只有一个线程写入文件：

import threading
file_lock = threading.Lock()

def save_url(str_url):
    with file_lock:
        with open("list.txt", "a") as save:
            save.write(f"{str_url}\n")

将原始代码中的文件写入部分替换为save_url(str_url)
3. 释放IMAP连接：在try...finally块中关闭连接：

try:
    # 原有邮件处理逻辑
finally:
    mail.close()
    mail.logout()

4. 复用URLExtract实例：将extractor = URLExtract()移到函数外部，避免重复初始化

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