第八站练习答案

返回纵览 ｜ 专题一 ｜ 专题二

测试 settings：

TASKS = {
    "default": {"BACKEND": "django.tasks.backends.dummy.DummyBackend"},
}

from django.tasks import default_task_backend
from django.test import SimpleTestCase, override_settings
from tickets.tasks import send_ticket_notice


@override_settings(
    TASKS={"default": {"BACKEND": "django.tasks.backends.dummy.DummyBackend"}}
)
class NoticeTaskTests(SimpleTestCase):
    def tearDown(self):
        default_task_backend.clear()

    def test_enqueue_contract(self):
        send_ticket_notice.enqueue(ticket_id=7, recipient="learner@example.com")
        self.assertEqual(len(default_task_backend.results), 1)

专题二核对

• 为 event_id 建唯一约束；两个 worker 同时 get_or_create() 时捕获唯一冲突并重新读取，完成者写入 sent_at。
• 把同一 event_id 传给邮件供应商作为幂等键；不支持时记录发送尝试和人工补偿边界，不能声称 exactly-once。
• ASGI 存活只验证进程事件循环；就绪验证必要依赖。Task worker 则分别验证进程心跳、队列连接与积压，避免数据库故障直接触发无限重启。

深水边界

这只证明 Task 的入队合同，不证明第三方 backend、worker、网络或邮件服务。生产必须分别做 backend 集成测试和 worker 端到端演练。

任务幂等可建立 Notification(event_id unique, sent_at)：先按 event_id 获取/创建记录，已完成则返回；发送与状态写回仍可能形成外部副作用窗口，需要供应商幂等键或可接受重复的业务设计。

开放练习（不提供答案）

定义任务队列不可用时的 HTTP 行为：失败请求、提交业务但延迟通知、或写 outbox。为选择写出用户影响和恢复证据。

继续第九站

加码题解：异步边界与幂等任务

def claim_job(job, now):
    if job.status in {'DONE', 'RUNNING'}:
        return False
    job.status = 'RUNNING'
    job.started_at = now
    job.save(update_fields=['status', 'started_at'])
    return True

关键推理

状态转换在数据库中留下可审计事实；重复投递不会重复领取已完成任务。

可替代方案

对外部支付使用供应商幂等键，对内部任务使用唯一事件键和 outbox。

典型错误

只在 worker 内存里记录已处理 ID，进程重启或多副本后立即失效。

复盘时请把状态码/事务/权限/资源清理中的至少一项写成自动化测试，而不是只截图成功页面。