跳转至

Celery

适用于通过 Celery（Python 3.10+）进行后台邮件投递的集成指南。邮件从 worker 经 Mailexam SMTP 发送 — 适用于注册、通知及 CI，且不阻塞 HTTP 请求。

可与 Django、FastAPI 或任何其他应用配合：Web 层入队任务，worker 发送邮件。

前置条件

  • Mailexam 账户，以及已创建并具备 SMTP 凭据的项目。
  • 消息代理（本示例使用 Redis）。
  • Python 3.10+ 及虚拟环境。

从项目的欢迎邮件（或控制台）中复制：

  • YOUR_LOGIN — SMTP 登录名（例如 xxxxx）；
  • YOUR_PASSWORD — SMTP 密码（与登录名成对的唯一凭据）；
  • 主机 — YOUR_LOGIN.mailexam.cn（与登录名一致）。

1. 依赖

python3 -m venv .venv
source .venv/bin/activate
pip install celery redis python-dotenv

requirements.txt：

celery>=5.4
redis>=5.0
python-dotenv>=1.0

本地运行 Redis（Docker）：

docker run -d --name redis -p 6379:6379 redis:7-alpine

2. 环境变量

项目根目录的 .env 文件：

CELERY_BROKER_URL=redis://localhost:6379/0

MAILEXAM_LOGIN=YOUR_LOGIN
MAILEXAM_PASSWORD=YOUR_PASSWORD
MAILEXAM_PORT=587
MAIL_FROM=noreply@example.test

SMTP 主机：{MAILEXAM_LOGIN}.mailexam.cn。

3. 邮件发送模块

与 Flask 相同的 smtplib 方式 — 项目根目录的 mail.py 文件。

4. Celery 应用与任务

# celery_app.py
import os

from celery import Celery
from dotenv import load_dotenv

load_dotenv()

app = Celery(
    "mailexam",
    broker=os.environ.get("CELERY_BROKER_URL", "redis://localhost:6379/0"),
    include=["tasks"],
)

# tasks.py
from celery_app import app
from mail import send_test


@app.task(name="mail.send_test")
def send_test_email(
    to: str = "user@example.test",
    subject: str = "Celery + Mailexam",
    body: str = "来自 Celery 的 Mailexam 测试",
) -> dict:
    send_test(to=to, subject=subject, body=body)
    return {"status": "ok", "to": to}

5. 运行 worker 与验证

终端 1 — worker：

celery -A celery_app worker --loglevel=info

终端 2 — 入队任务：

python -c "from tasks import send_test_email; r = send_test_email.delay(); print(r.id)"

或通过 shell：

celery -A celery_app shell

from tasks import send_test_email
send_test_email.delay(
    to="user@example.test",
    subject="测试",
    body="来自队列的问候",
)

邮件将出现在 Mailexam 控制台 → 您的项目 → 收件箱。

Django 集成

若邮件已在 Django 中配置，任务可调用 send_mail：

from celery import shared_task
from django.core.mail import send_mail


@shared_task
def send_welcome_email(to: str) -> None:
    send_mail(
        "Welcome",
        "通过 Celery + Django 的 Mailexam 测试",
        None,
        [to],
    )

在 Django 项目中运行 worker：

celery -A config worker --loglevel=info

（config — 含 Celery 应用的模块，通常为 config/celery.py。）

6. 本地开发与 CI

环境 建议
local Redis + 带 Mailexam 凭据的 .env
CI 流水线中的 Redis 服务，密钥 MAILEXAM_LOGIN、MAILEXAM_PASSWORD

GitLab CI 变量示例：

variables:
  CELERY_BROKER_URL: redis://redis:6379/0
  MAILEXAM_LOGIN: $MAILEXAM_LOGIN
  MAILEXAM_PASSWORD: $MAILEXAM_PASSWORD
  MAILEXAM_PORT: "587"
  MAIL_FROM: "noreply@example.test"

无需 broker 和 SMTP 的测试：

app.conf.task_always_eager = True

任务在同一进程中同步运行（便于单元测试；除非 mock send_test，否则邮件仍会进入 Mailexam）。

流水线发送后，通过 Mailexam API 验证投递。

7. 常见问题

任务停留在 PENDING 状态

  • worker 是否在运行：celery -A celery_app worker --loglevel=info？
  • worker 与客户端的 CELERY_BROKER_URL 是否一致，Redis 是否可达。

worker 日志中的 SMTP 错误

  • 主机 {login}.mailexam.cn，登录名和密码 — 同一项目的邮件成对凭据。
  • worker 仅在 celery_app.py 中调用 load_dotenv() 时才会继承 .env（如示例）。

控制台中看不到邮件

  • 查看 worker .env 中凭据对应的同一 Mailexam 项目收件箱。
  • 检查任务结果：celery -A celery_app result <task_id>。

参见