Python邮件验证迭代疑似停滞？如何优化代码运行速度？

问题2：代码速度优化方案 #

原代码是单线程串行处理，IO等待时间占比极高，以下是针对性优化：

1. 并发处理（核心优化）

使用线程池处理IO密集型任务，大幅提升效率：

from concurrent.futures import ThreadPoolExecutor

def verify_email(email):
    # 将原循环内的验证逻辑封装成独立函数
    addressToVerify = str(email)
    # ... 验证代码（含超时设置）
    return (email, result)

# 根据网络情况调整max_workers（建议20-50）
with ThreadPoolExecutor(max_workers=30) as executor:
    results = list(executor.map(verify_email, data))

2. 缓存MX记录

同一个域名的MX记录无需重复查询，用字典缓存：

mx_cache = {}

def get_mx_record(domain):
    if domain in mx_cache:
        return mx_cache[domain]
    try:
        records = dns.resolver.query(domain, 'MX', lifetime=5)
        mx_record = str(records[0].exchange)
        mx_cache[domain] = mx_record
        return mx_record
    except:
        mx_cache[domain] = None
        return None

3. 预编译正则表达式

避免每次循环编译正则，提前编译一次：

email_regex = re.compile(r'^[_a-z0-9-]+(\.[_a-z0-9-]+)*@[a-z0-9-]+(\.[a-z0-9-]+)*(\.[a-z]{2,})$')
# 后续使用：match = email_regex.match(addressToVerify.lower())

4. 优化结果存储

用元组列表替代字典列表，减少内存开销：

# 原：df.append({n:'Bad Syntax'})
# 改：results.append( (n, 'Bad Syntax') )
# 最后转JSON：json.dump(results, fout)

5. 其他小优化

• 避免裸except：捕获具体异常（如dns.resolver.NXDOMAIN、smtplib.SMTPException、socket.timeout等），防止意外中断。
• 批量读取Excel：若文件过大，用pd.read_excel(chunksize=1000)分批处理，降低内存占用。
• 关闭SMTP调试：确保server.set_debuglevel(0)，减少不必要输出。

优化后的完整代码示例 #

import re
import dns.resolver
import socket
import smtplib
import pandas as pd
import json
from concurrent.futures import ThreadPoolExecutor

# 预编译正则表达式
email_regex = re.compile(r'^[_a-z0-9-]+(\.[_a-z0-9-]+)*@[a-z0-9-]+(\.[a-z0-9-]+)*(\.[a-z]{2,})$')
# MX记录缓存字典
mx_cache = {}

def get_mx_record(domain):
    if domain in mx_cache:
        return mx_cache[domain]
    try:
        records = dns.resolver.query(domain, 'MX', lifetime=5)
        mx_record = str(records[0].exchange)
        mx_cache[domain] = mx_record
        return mx_record
    except Exception as e:
        mx_cache[domain] = None
        return None

def verify_email(email):
    addressToVerify = str(email)
    match = email_regex.match(addressToVerify.lower())
    
    if not match:
        return (email, 'Bad Syntax')
    
    try:
        domain = addressToVerify.split('@')[1]
    except IndexError:
        return (email, 'Bad Syntax')
    
    mx_record = get_mx_record(domain)
    if not mx_record:
        return (email, 'Server Unreachable')
    
    try:
        host = socket.gethostname()
        server = smtplib.SMTP(timeout=10)
        server.set_debuglevel(0)
        server.connect(mx_record)
        server.helo(host)
        server.mail('me@domain.com')
        code, message = server.rcpt(addressToVerify)
        server.quit()
        
        if code == 250:
            return (email, 'Success')
        else:
            return (email, 'Bad')
    except Exception as e:
        return (email, 'Server Unreachable')

if __name__ == '__main__':
    # 读取邮箱数据
    data = pd.read_excel(r"C:\Users\win10\Downloads\USA Emails.xlsx").iloc[:,0].tolist()
    # 选取15万条进行验证
    data_sample = data[:150000]
    
    # 并发处理邮箱验证
    with ThreadPoolExecutor(max_workers=30) as executor:
        results = list(executor.map(verify_email, data_sample))
    
    # 保存验证结果
    with open('dataaa.json', 'w', encoding='utf-8') as fout:
        json.dump(results, fout, ensure_ascii=False)
    
    print(f"处理完成，共验证 {len(results)} 条邮箱")

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