如何为SMTP连接设置超时以终止耗时过长的邮箱验证函数

我通过解析DNS+SMTP对话的方式验证邮箱有效性，代码如下：

def checkEmailState(email):
    fromAddress = 'mislam162013@bscse.uiu.ac.bd'

    # Get domain for DNS lookup
    splitAddress = email.split('@')
    domain = str(splitAddress[1])

    try:
        # MX record lookup
        records = dns.resolver.resolve(domain, 'MX')
        mxRecord = records[0].exchange
        mxRecord = str(mxRecord)

        # SMTP lib setup (use debug level for full output)
        server = smtplib.SMTP()
        server.set_debuglevel(0)

        # SMTP Conversation
        server.connect(mxRecord)
        server.helo(server.local_hostname)  ### server.local_hostname(Get local server hostname)
        server.mail(fromAddress)
        code, message = server.rcpt(str(email))
        server.quit()
    except smtplib.SMTPServerDisconnected:
        return 'Success'
    except TimeoutError:
        return 'Success'
    except dns.resolver.NoAnswer:
        return 'Bad'
    except UnicodeError:
        return 'Bad'
    except:
        return 'Bad'

但执行server.connect(mxRecord)时耗时极久且无错误返回，尝试用socket设置超时未生效，需要实现函数执行超过设定时长就终止并返回结果。

方法1：直接使用smtplib内置超时参数 #

smtplib.SMTP本身支持设置超时时间，在初始化或connect调用时传入timeout参数即可，该参数会控制连接、SMTP命令等操作的超时阈值：

import dns.resolver
import smtplib

def checkEmailState(email, timeout=10):
    fromAddress = 'mislam162013@bscse.uiu.ac.bd'

    splitAddress = email.split('@')
    if len(splitAddress) != 2:
        return 'Bad'
    domain = splitAddress[1]

    try:
        # MX记录查询
        records = dns.resolver.resolve(domain, 'MX')
        mxRecord = str(records[0].exchange)

        # 初始化SMTP客户端时设置超时（单位：秒）
        server = smtplib.SMTP(timeout=timeout)
        server.set_debuglevel(0)

        server.connect(mxRecord)
        server.helo(server.local_hostname)
        server.mail(fromAddress)
        code, message = server.rcpt(email)
        server.quit()

        # 根据rcpt返回码判断邮箱有效性，250表示存在
        return 'Success' if code == 250 else 'Bad'
    except smtplib.SMTPServerDisconnected:
        return 'Success'
    except (TimeoutError, dns.resolver.NoAnswer, UnicodeError):
        return 'Bad'
    except Exception:
        # 捕获其他未预期异常，避免裸except
        return 'Bad'

若连接或后续SMTP操作超过设定时长未响应，会自动抛出TimeoutError，进入异常分支返回Bad。

方法2：用线程实现整体函数超时控制 #

如果方法1无法解决某些极端场景（比如MX服务器故意静默不响应），可以用线程封装检查逻辑，通过join(timeout)强制控制整体执行时长：

import dns.resolver
import smtplib
import threading

def _email_check_worker(email, result):
    fromAddress = 'mislam162013@bscse.uiu.ac.bd'
    try:
        splitAddress = email.split('@')
        if len(splitAddress) != 2:
            result.append('Bad')
            return
        domain = splitAddress[1]

        records = dns.resolver.resolve(domain, 'MX')
        mxRecord = str(records[0].exchange)

        server = smtplib.SMTP(timeout=5)
        server.set_debuglevel(0)
        server.connect(mxRecord)
        server.helo(server.local_hostname)
        server.mail(fromAddress)
        code, message = server.rcpt(email)
        server.quit()

        result.append('Success' if code == 250 else 'Bad')
    except smtplib.SMTPServerDisconnected:
        result.append('Success')
    except (TimeoutError, dns.resolver.NoAnswer, UnicodeError):
        result.append('Bad')
    except Exception:
        result.append('Bad')

def checkEmailState(email, timeout=10):
    result = []
    # 创建并启动检查线程
    worker_thread = threading.Thread(target=_email_check_worker, args=(email, result))
    worker_thread.start()
    # 等待线程执行，超时则终止等待
    worker_thread.join(timeout)
    
    # 线程仍存活说明超时
    if worker_thread.is_alive():
        return 'Bad'
    # 返回线程执行结果
    return result[0] if result else 'Bad'

说明： #

• 把邮箱检查逻辑放到子线程中执行，主线程通过join(timeout)等待结果
• 超过设定时间后，若线程仍在运行，直接判定为超时返回Bad
• 用列表result传递线程执行结果（线程无法直接返回值）

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