如何高效批量发送带自定义内容与附件的邮件？（需演示SMTP多线程及Gmail API两种方案）

嘿，针对你要发送数千封带个性化姓名和附件邮件的需求，我整理了两种高效落地的方案——多线程SMTP和Gmail API批量发送，这俩都能帮你大幅提升发送效率，避免单线程一封一封发的低效问题。

方案一：多线程SMTP发送（适合快速落地，兼容多数邮箱） #

单线程SMTP之所以慢，是因为每发一封都要等上一封完成。用多线程并行处理，能同时发起多个SMTP连接（注意别超过服务商的并发限制，比如Gmail建议并发数控制在5-10以内），直接把发送效率拉上来。

我基于你提供的代码框架，改成了多线程版本，核心用concurrent.futures.ThreadPoolExecutor来管理线程池，同时优化了邮件生成和异常处理：

from os import path
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.application import MIMEApplication
from concurrent.futures import ThreadPoolExecutor, as_completed
from time import perf_counter

class EmailSender:
    def __init__(self, username, password, smtp_server, smtp_port, data, subject_template, text_template, html_template, attachments):
        self.username = username
        self.password = password
        self.smtp_server = smtp_server
        self.smtp_port = smtp_port
        self.data = data  # 格式：[{"name": "张三", "email": "xxx@xxx.com"}, ...]
        self.subject_template = subject_template
        self.text_template = text_template
        self.html_template = html_template
        self.attachments = attachments  # 格式：["path/to/file1.pdf", "path/to/file2.docx"]
        self.failed_emails = []

    def _generate_email(self, receiver_info):
        # 创建邮件实例
        msg = MIMEMultipart()
        msg['From'] = self.username
        msg['To'] = receiver_info['email']
        # 替换模板变量
        msg['Subject'] = self.subject_template.format(name=receiver_info['name'])
        # 添加文本和HTML内容
        msg.attach(MIMEText(self.text_template.format(name=receiver_info['name']), 'plain'))
        if self.html_template:
            msg.attach(MIMEText(self.html_template.format(name=receiver_info['name']), 'html'))
        # 添加附件
        for attachment_path in self.attachments:
            if path.exists(attachment_path):
                with open(attachment_path, 'rb') as f:
                    part = MIMEApplication(f.read(), Name=path.basename(attachment_path))
                    part['Content-Disposition'] = f'attachment; filename="{path.basename(attachment_path)}"'
                    msg.attach(part)
        return msg.as_string()

    def _send_single_email(self, receiver_info):
        try:
            # 每个线程单独建立SMTP连接（避免线程安全问题）
            with smtplib.SMTP_SSL(self.smtp_server, self.smtp_port) as server:
                server.login(self.username, self.password)
                email_content = self._generate_email(receiver_info)
                server.sendmail(self.username, receiver_info['email'], email_content)
                print(f"成功发送给 {receiver_info['name']} <{receiver_info['email']}>")
        except Exception as e:
            print(f"发送失败给 {receiver_info['name']} <{receiver_info['email']}>：{str(e)}")
            self.failed_emails.append(receiver_info)

    def send_batch(self, max_workers=8):
        start_time = perf_counter()
        # 用线程池并行发送
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = [executor.submit(self._send_single_email, info) for info in self.data]
            # 等待所有任务完成
            for future in as_completed(futures):
                pass
        end_time = perf_counter()
        print(f"\n批量发送完成！耗时：{end_time - start_time:.2f}秒")
        print(f"失败邮件数量：{len(self.failed_emails)}")
        if self.failed_emails:
            print("失败列表：", [info['email'] for info in self.failed_emails])

# 示例使用
if __name__ == "__main__":
    # 配置信息
    USERNAME = "your_email@gmail.com"
    PASSWORD = "your_app_password"  # Gmail需要用应用专用密码，开启两步验证后生成
    SMTP_SERVER = "smtp.gmail.com"
    SMTP_PORT = 465
    # 收件人数据
    RECIPIENTS = [
        {"name": "Alice", "email": "alice@example.com"},
        {"name": "Bob", "email": "bob@example.com"},
        # 更多收件人...
    ]
    # 邮件模板
    SUBJECT_TPL = "嗨，{name}！这是你的专属邮件"
    TEXT_TPL = "你好{name}，\n\n这是为你定制的内容，附件请查收。\n\n祝好！"
    HTML_TPL = "<h3>你好{name}</h3><p>这是为你定制的内容，附件请查收。</p>"
    # 附件路径
    ATTACHMENTS = ["report.pdf", "guide.docx"]

    sender = EmailSender(USERNAME, PASSWORD, SMTP_SERVER, SMTP_PORT, RECIPIENTS, SUBJECT_TPL, TEXT_TPL, HTML_TPL, ATTACHMENTS)
    sender.send_batch(max_workers=8)

关键注意点： #

• 每个线程单独建立SMTP连接，避免多线程共享连接导致的线程安全问题
• 控制max_workers的数量，别太激进（比如Gmail建议不超过10），不然容易被判定为垃圾邮件或者临时限制
• 一定要用SMTP_SSL加密连接，避免账号信息泄露
• 记录失败的邮件，方便后续重试

方案二：Gmail API批量发送（更合规，大规模发送首选） #

如果是用Gmail发送，官方的Gmail API支持批量请求，能把多封邮件打包成一个HTTP请求发送，比SMTP更高效，也更符合Google的规则，不容易触发限制。

前置准备： #

1. 去Google Cloud Console创建项目，启用Gmail API
2. 生成OAuth2客户端凭据（个人账号用这个），或者服务账号（企业域账号用）
3. 安装依赖：pip install google-api-python-client google-auth-httplib2 google-auth-oauthlib

下面是批量发送的实现代码：

import base64
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.application import MIMEApplication
from os import path
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
import pickle
from time import perf_counter

# Gmail API权限范围
SCOPES = ['https://www.googleapis.com/auth/gmail.send']

class GmailAPISender:
    def __init__(self, credentials_path, token_path, data, subject_template, text_template, html_template, attachments):
        self.credentials_path = credentials_path
        self.token_path = token_path
        self.data = data
        self.subject_template = subject_template
        self.text_template = text_template
        self.html_template = html_template
        self.attachments = attachments
        self.service = self._get_gmail_service()
        self.failed_emails = []

    def _get_gmail_service(self):
        """获取Gmail API服务实例"""
        creds = None
        # 加载已保存的token
        if path.exists(self.token_path):
            with open(self.token_path, 'rb') as token:
                creds = pickle.load(token)
        # 如果没有有效凭据，重新登录
        if not creds or not creds.valid:
            if creds and creds.expired and creds.refresh_token:
                creds.refresh(Request())
            else:
                flow = InstalledAppFlow.from_client_secrets_file(self.credentials_path, SCOPES)
                creds = flow.run_local_server(port=0)
            # 保存token供下次使用
            with open(self.token_path, 'wb') as token:
                pickle.dump(creds, token)
        return build('gmail', 'v1', credentials=creds)

    def _create_message(self, receiver_info):
        """生成符合Gmail API要求的邮件消息"""
        msg = MIMEMultipart()
        msg['From'] = 'me'  # Gmail API用'me'代表当前授权用户
        msg['To'] = receiver_info['email']
        msg['Subject'] = self.subject_template.format(name=receiver_info['name'])
        msg.attach(MIMEText(self.text_template.format(name=receiver_info['name']), 'plain'))
        if self.html_template:
            msg.attach(MIMEText(self.html_template.format(name=receiver_info['name']), 'html'))
        # 添加附件
        for attachment_path in self.attachments:
            if path.exists(attachment_path):
                with open(attachment_path, 'rb') as f:
                    part = MIMEApplication(f.read(), Name=path.basename(attachment_path))
                    part['Content-Disposition'] = f'attachment; filename="{path.basename(attachment_path)}"'
                    msg.attach(part)
        # 转成base64编码，符合Gmail API要求
        raw = base64.urlsafe_b64encode(msg.as_bytes()).decode()
        return {'raw': raw}

    def send_batch(self, batch_size=50):
        """批量发送邮件，每batch_size封打包一次"""
        start_time = perf_counter()
        total = len(self.data)
        for i in range(0, total, batch_size):
            batch = self.service.new_batch_http_request(callback=self._batch_callback)
            batch_recipients = self.data[i:i+batch_size]
            for info in batch_recipients:
                try:
                    message = self._create_message(info)
                    batch.add(self.service.users().messages().send(userId='me', body=message))
                except Exception as e:
                    print(f"添加邮件失败 {info['name']} <{info['email']}>：{str(e)}")
                    self.failed_emails.append(info)
            # 执行批量请求
            try:
                batch.execute()
                print(f"完成第 {i//batch_size + 1} 批发送，共 {len(batch_recipients)} 封")
            except HttpError as e:
                print(f"批量请求失败：{str(e)}")
        end_time = perf_counter()
        print(f"\n批量发送完成！耗时：{end_time - start_time:.2f}秒")
        print(f"失败邮件数量：{len(self.failed_emails)}")
        if self.failed_emails:
            print("失败列表：", [info['email'] for info in self.failed_emails])

    def _batch_callback(self, request_id, response, exception):
        """批量请求的回调函数，处理每个请求的结果"""
        if exception:
            print(f"发送失败：{str(exception)}")
            self.failed_emails.append({"email": response.get('to') if response else "未知邮箱"})
        else:
            print(f"邮件发送成功，消息ID：{response['id']}")

# 示例使用
if __name__ == "__main__":
    # 凭据路径（从Google Cloud Console下载的json文件）
    CREDENTIALS_PATH = 'credentials.json'
    TOKEN_PATH = 'token.pickle'
    # 收件人数据
    RECIPIENTS = [
        {"name": "Alice", "email": "alice@example.com"},
        {"name": "Bob", "email": "bob@example.com"},
        # 更多收件人...
    ]
    # 邮件模板
    SUBJECT_TPL = "嗨，{name}！这是你的专属邮件"
    TEXT_TPL = "你好{name}，\n\n这是为你定制的内容，附件请查收。\n\n祝好！"
    HTML_TPL = "<h3>你好{name}</h3><p>这是为你定制的内容，附件请查收。</p>"
    # 附件路径
    ATTACHMENTS = ["report.pdf", "guide.docx"]

    sender = GmailAPISender(CREDENTIALS_PATH, TOKEN_PATH, RECIPIENTS, SUBJECT_TPL, TEXT_TPL, HTML_TPL, ATTACHMENTS)
    sender.send_batch(batch_size=50)

关键注意点： #

• 批量大小batch_size建议设为50-100，Google的API对批量请求的数量有上限
• 第一次运行会弹出浏览器授权，之后会保存token，下次直接用
• 一定要遵守Gmail的发送限制：普通账号每天最多500封，企业域账号最多2000封
• 回调函数可以精准处理每一封邮件的发送结果，方便排查问题

通用优化建议 #

不管用哪种方案，这些技巧都能帮你更高效、安全地发送：

• 加入速率限制：比如每发送N封就暂停几秒，避免触发服务商的反垃圾机制
• 做好模板变量替换：确保每个收件人的内容都是个性化的，减少被判定为垃圾邮件的概率
• 实现重试机制：对发送失败的邮件，隔一段时间后重新发送
• 记录发送日志：保存每封邮件的发送状态，方便后续排查

备注：内容来源于stack exchange，提问作者makalidap