您现在的位置：程式碼之家 >> 編程語言 > >> 更多程式語言 >> 更多程式設計知識

Flask-Mail 異步發送郵件實戰：從同步阻塞到線程池優化

編輯：更多程式設計知識

Web 應用發送郵件是典型 I/O 密集型操作。同步調用 mail.send() 會阻塞工作進程數秒，嚴重降低吞吐量。本文演示 **Flask-Mail 基礎配置、模板化郵件、同步阻塞問題、線程異步化、裝飾器封裝、線程池進階** 完整演進路徑。

Flask-Mail 基礎配置

# config.py
MAIL_SERVER = 'smtp.googlemail.com'
MAIL_PORT = 465
MAIL_USE_SSL = True
MAIL_USERNAME = 'your@gmail.com'
MAIL_PASSWORD = 'your-password'

ADMINS = ['your@gmail.com']

# app/__init__.py
from flask import Flask
from flask_mail import Mail

app = Flask(__name__)
app.config.from_object('config')
mail = Mail(app)

郵件模板化：分離邏輯與表現

文本版 templates/follower_email.txt：

Dear {{ user.nickname }},

{{ follower.nickname }} is now following you.
Profile: {{ url_for('user', nickname=follower.nickname, _external=True) }}

Regards,
The microblog admin

HTML 版 templates/follower_email.html：

<p>Dear {{ user.nickname }},</p>
<p><a href="{{ url_for('user', nickname=follower.nickname, _external=True) }}">{{ follower.nickname }}</a> is now following you.</p>
<table><tr><td><img src="{{ follower.avatar(50) }}"></td><td>
  <a href="{{ url_for('user', nickname=follower.nickname, _external=True) }}">{{ follower.nickname }}</a><br>
  {{ follower.about_me }}
</td></tr></table>
<p>Regards,</p><p>The <code>microblog</code> admin</p>

關鍵點：url_for(..., _external=True) 生成絕對 URL（含域名），郵件客戶端無上下文必須用絕對鏈接。

同步發送視圖（阻塞問題）

from flask_mail import Message
from app import mail
from config import ADMINS

def send_email(subject, sender, recipients, text_body, html_body):
    msg = Message(subject, sender=sender, recipients=recipients)
    msg.body = text_body
    msg.html = html_body
    mail.send(msg)  # 阻塞直到 SMTP 交付完成

# 視圖調用
@app.route('/follow/')
@login_required
def follow(nickname):
    user = User.query.filter_by(nickname=nickname).first()
    # ...
    send_email(
        f"[microblog] {follower.nickname} is now following you!",
        ADMINS[0], [user.email],
        render_template('follower_email.txt', user=user, follower=follower),
        render_template('follower_email.html', user=user, follower=follower)
    )
    return redirect(url_for('user', nickname=nickname))

點擊「關注」→ 瀏覽器卡頓 2-3 秒 → 頁面跳轉。**不可接受。**

線程異步化：非阻塞發送


from threading import Thread
from flask import current_app
from flask_mail import Message
from app import mail

def send_async_email(app, msg):
    with app.app_context():  # 關鍵：異步線程需手動推送應用上下文
        mail.send(msg)

def send_email(subject, sender, recipients, text_body, html_body):
    msg = Message(subject, sender=sender, recipients=recipients)
    msg.body = text_body
    msg.html = html_body
    thr = Thread(target=send_async_email, args=[current_app._get_current_object(), msg])
    thr.start()
    return thr  # 可選：返回線程對象供監控/join

瀏覽器即時響應，郵件在後台線程發送。

裝飾器封裝：通用異步執行器


# decorators.py
from threading import Thread
from functools import wraps

def async(f):
    @wraps(f)
    def wrapper(*args, **kwargs):
        thr = Thread(target=f, args=args, kwargs=kwargs)
        thr.start()
        return thr
    return wrapper

# emails.py
from decorators import async
from flask_mail import Message
from app import mail

@async
def send_async_email(msg):
    with current_app.app_context():
        mail.send(msg)

def send_email(subject, sender, recipients, text_body, html_body):
    msg = Message(subject, sender=sender, recipients=recipients)
    msg.body = text_body
    msg.html = html_body
    send_async_email(msg)  # 即刻返回，後台執行

通用 @async 裝飾器可復用於任何耗時 I/O 操作（推送、Webhook、報表生成等）。

進階：線程池控制併發

無限創建線程會耗盡資源。生產環境建議使用 **線程池**：


from concurrent.futures import ThreadPoolExecutor
from functools import wraps

# 應用啟動時創建全局池（如 10 線程）
executor = ThreadPoolExecutor(max_workers=10)

def async_pool(f):
    @wraps(f)
    def wrapper(*args, **kwargs):
        return executor.submit(f, *args, **kwargs)  # 返回 Future
    return wrapper

# 使用
@async_pool
def send_async_email(msg):
    with current_app.app_context():
        mail.send(msg)

優勢：