Django多线程调用Gmail SMTP发件时连接意外断开问题求助

我在Django框架下采用多线程方式，通过Gmail SMTP批量发送独立邮件以提升效率，但遇到连接被断开的问题。发送约50-55封邮件后，会触发如下错误：

smtplib.SMTPServerDisconnected: Connection unexpectedly closed

由于需要记录每封邮件的发送成功/失败状态到数据库用于审计，我无法使用Django的send_mass_email方法。

代码实现 #

Threading类 #

class Threading(Thread):
def __init__(self, emails):
    self.emails = emails
    Threading.thread(self)

def thread(self):
    queue = Queue()

    #4 worker threads
    for x in range(4):

        worker = SenderWorker(queue)
        worker.daemon = True
        worker.start()

    
    for email in self.emails:
        queue.put( email )

    queue.join()

SenderWorker类 #

class SenderWorker(Thread):

def __init__(self, queue):
    Thread.__init__(self)
    self.queue = queue

def run(self):
    start = datetime.now()
    while True:
        email = self.queue.get()
        try:
            Email.sendEmail( email ) #method for sending and logging a single email sent
        finally:
            self.queue.task_done()

SMTP配置 #

EMAIL_BACKEND = 'django.core.mail.backends.smtp.EmailBackend'
EMAIL_USE_TLS = True
EMAIL_HOST = 'smtp.gmail.com'
EMAIL_PORT = 587
EMAIL_HOST_USER = 'anon@anonymous.com'
SERVER_EMAIL = 'anon@anonymous.com'
EMAIL_HOST_PASSWORD = '*******'
DEFAULT_FROM_EMAIL = EMAIL_HOST_USER

已尝试方法 #

• 调整线程数量
• 切换SSL/TLS端口及认证方式
• 减少单次发送邮件数量/分批发送

问题仍未解决，我未找到Gmail SMTP关于每秒发件数或每秒连接数的限制，希望找到无需添加休眠时间的解决方案。

1. 让Worker线程复用SMTP连接 #

当前每发送一封邮件可能都会新建SMTP连接，频繁创建连接是触发Gmail断开的核心原因。修改SenderWorker类，让每个线程保持一个持久连接，连接断开时自动重试重建：

class SenderWorker(Thread):
    def __init__(self, queue):
        Thread.__init__(self)
        self.queue = queue
        self.smtp_conn = None
        # 从配置中读取SMTP参数
        self.smtp_host = 'smtp.gmail.com'
        self.smtp_port = 587
        self.smtp_user = 'anon@anonymous.com'
        self.smtp_pass = '*******'

    def _init_smtp_conn(self):
        """初始化SMTP连接"""
        try:
            conn = smtplib.SMTP(self.smtp_host, self.smtp_port)
            conn.starttls()
            conn.login(self.smtp_user, self.smtp_pass)
            return conn
        except Exception as e:
            print(f"SMTP连接初始化失败: {str(e)}")
            return None

    def run(self):
        self.smtp_conn = self._init_smtp_conn()
        if not self.smtp_conn:
            return

        while True:
            email = self.queue.get()
            try:
                # 复用现有连接发送邮件，需修改sendEmail方法支持传入连接
                Email.sendEmail(email, smtp_conn=self.smtp_conn)
            except smtplib.SMTPServerDisconnected:
                # 连接断开，重建后重试发送
                self.smtp_conn = self._init_smtp_conn()
                if self.smtp_conn:
                    Email.sendEmail(email, smtp_conn=self.smtp_conn)
                else:
                    # 记录邮件发送失败状态到数据库
                    print(f"无法重建连接，邮件{email}发送失败")
            except Exception as e:
                # 处理其他发送错误，记录失败状态
                print(f"邮件{email}发送失败: {str(e)}")
            finally:
                self.queue.task_done()

    def __del__(self):
        # 线程销毁时关闭SMTP连接
        if self.smtp_conn:
            try:
                self.smtp_conn.quit()
            except:
                pass

同时修改Email.sendEmail方法，支持传入SMTP连接：

@classmethod
def sendEmail(cls, email, smtp_conn=None):
    # 保留原有的邮件构建逻辑
    msg = cls._build_email_msg(email)  # 假设这是你构建邮件的内部方法

    if smtp_conn:
        try:
            smtp_conn.send_message(msg)
            # 记录邮件发送成功状态到数据库
            return True
        except Exception as e:
            # 记录失败状态
            cls._log_send_status(email, success=False, error=str(e))
            return False
    # 兼容无连接传入的情况（可选）
    else:
        with smtplib.SMTP('smtp.gmail.com', 587) as conn:
            conn.starttls()
            conn.login('anon@anonymous.com', '*******')
            conn.send_message(msg)
            cls._log_send_status(email, success=True)
            return True

2. 实现SMTP连接池 #

可以封装一个连接池类，让所有Worker线程从池中获取、归还连接，避免每个线程单独维护连接。示例思路：

from queue import Queue
import smtplib

class SMTPConnectionPool:
    def __init__(self, host, port, user, password, pool_size=4):
        self.host = host
        self.port = port
        self.user = user
        self.password = password
        self.pool = Queue(maxsize=pool_size)
        # 初始化连接池
        for _ in range(pool_size):
            conn = self._create_conn()
            if conn:
                self.pool.put(conn)

    def _create_conn(self):
        try:
            conn = smtplib.SMTP(self.host, self.port)
            conn.starttls()
            conn.login(self.user, self.password)
            return conn
        except Exception as e:
            print(f"创建连接失败: {str(e)}")
            return None

    def get_conn(self):
        # 如果池为空，尝试新建连接
        if self.pool.empty():
            conn = self._create_conn()
            if conn:
                return conn
            # 新建失败则等待现有连接归还
            return self.pool.get()
        return self.pool.get()

    def return_conn(self, conn):
        try:
            # 检查连接是否可用，不可用则销毁重建
            conn.noop()
            self.pool.put(conn)
        except smtplib.SMTPServerDisconnected:
            new_conn = self._create_conn()
            if new_conn:
                self.pool.put(new_conn)

然后在Worker线程中使用连接池：

# 初始化连接池（全局或线程外初始化）
smtp_pool = SMTPConnectionPool(
    host='smtp.gmail.com',
    port=587,
    user='anon@anonymous.com',
    password='*******',
    pool_size=4
)

class SenderWorker(Thread):
    def __init__(self, queue):
        Thread.__init__(self)
        self.queue = queue

    def run(self):
        while True:
            email = self.queue.get()
            conn = None
            try:
                conn = smtp_pool.get_conn()
                Email.sendEmail(email, smtp_conn=conn)
            except Exception as e:
                print(f"邮件{email}发送失败: {str(e)}")
                # 记录失败状态
            finally:
                if conn:
                    smtp_pool.return_conn(conn)
                self.queue.task_done()

3. 检查Gmail账号权限配置 #

• 如果开启了2FA，必须使用App密码而非账号密码登录SMTP
• 确保账号未触发Gmail的发送限制（普通账号日发送上限500封）
• 检查账号是否被标记为异常，可登录Gmail查看安全提醒

4. 优化任务批次 #

即使复用连接，单个连接长时间发送也可能被断开。可以让Worker每处理N封邮件后，重新获取/创建连接，比如每处理20封就更换一次连接，避免单个连接持续使用过久。

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