Django 中实现 scheduled payment 的最佳策略是什么？

2026-11-04技术老齐

结论

如果支付服务商支持 subscription、自动扣款和 webhook，应优先由支付服务商处理周期扣款。Django 负责保存订阅状态、处理 webhook，并在付款日前创建 payment link、发送提醒。

提醒、对账和补偿等后台任务可以这样安排：

使用 Celery Beat 定期扫描数据库，再由 Celery Worker 执行具体任务。
项目规模较小时，可以通过系统 cron 调用 Django management command。
没有精确时间要求时，每天扫描一次即可；如果需要精确到小时或分钟，再相应缩短扫描间隔。

不建议为每位客户创建延迟数周或数月的 Celery eta/countdown 任务。这类任务会跨越部署、队列重启和订阅变更，取消或调整起来也比较麻烦。可以把 next_billing_at 存入数据库，再定期扫描已经到期的数据。

职责划分

常见的处理流程是：

用户购买 subscription 后，在本地保存支付平台的 subscription ID、状态和下次账单时间。
定时任务扫描未来 10 天内到期的订阅，创建“发送 payment link”任务。
到达扣款时间后，由支付平台自动扣款。如果平台不支持自动订阅，则由后台任务调用支付 API。
Django 通过 webhook 接收付款成功、付款失败、退款和 subscription 取消等事件。
webhook 处理完成后，更新本地订阅状态和下一次账单时间。
定期执行对账任务，修复 webhook 丢失或处理失败造成的状态差异。

支付结果要以支付平台的 webhook 或主动查询结果为准。Celery 任务成功返回，只能说明任务已经执行，不能直接据此认定付款成功。

Celery 与 cron 怎么选

cron 适合任务较少、部署结构简单的项目。例如，每天运行一次 management command：

0 2 * * * /path/to/venv/bin/python /path/to/project/manage.py dispatch_billing_actions

cron 配置简单，也不需要消息队列。不过，重试、并发控制、任务状态和监控都要自行实现。如果部署在多台服务器上，还要保证只有一个实例运行该命令。

系统已经使用 Redis 或 RabbitMQ，或者需要失败重试、多个 worker 和任务监控时，可以选择 Celery：

Celery Beat -> 扫描到期订阅 -> 投递任务 -> Celery Worker -> 调用邮件或支付服务

Celery Beat 调度实例只能运行一个，否则同一周期任务可能被重复投递。即便只有一个 Beat，业务代码仍要支持幂等，因为 worker 重试和网络超时也可能导致重复执行。

数据模型示例

以下模型保存订阅的下一次账单时间，并通过唯一约束防止同一账期重复创建同类操作：

from django.db import models


class Subscription(models.Model):
    class Status(models.TextChoices):
        ACTIVE = "active", "Active"
        PAST_DUE = "past_due", "Past due"
        CANCELED = "canceled", "Canceled"

    customer_id = models.CharField(max_length=128)
    provider_subscription_id = models.CharField(
        max_length=128,
        unique=True,
        null=True,
        blank=True,
    )
    status = models.CharField(
        max_length=20,
        choices=Status.choices,
        default=Status.ACTIVE,
    )
    next_billing_at = models.DateTimeField(db_index=True)


class BillingAction(models.Model):
    class Kind(models.TextChoices):
        SEND_PAYMENT_LINK = "send_payment_link", "Send payment link"
        CHARGE = "charge", "Charge"

    class Status(models.TextChoices):
        PENDING = "pending", "Pending"
        PROCESSING = "processing", "Processing"
        SUCCEEDED = "succeeded", "Succeeded"
        FAILED = "failed", "Failed"

    subscription = models.ForeignKey(
        Subscription,
        on_delete=models.CASCADE,
        related_name="billing_actions",
    )
    kind = models.CharField(max_length=32, choices=Kind.choices)
    billing_at = models.DateTimeField()
    status = models.CharField(
        max_length=20,
        choices=Status.choices,
        default=Status.PENDING,
    )
    idempotency_key = models.CharField(max_length=128, unique=True)
    last_error = models.TextField(blank=True)
    created_at = models.DateTimeField(auto_now_add=True)

    class Meta:
        constraints = [
            models.UniqueConstraint(
                fields=["subscription", "kind", "billing_at"],
                name="unique_action_per_billing_cycle",
            ),
        ]

每个账期单独记录一条 BillingAction，可以区分不同账期，也便于审计、重试和排查重复扣款。只在 Subscription 上保存一个 reminder_sent 布尔值，无法清楚记录这些信息。

定期扫描并创建任务

以下示例使用 Celery。扫描任务可以每天运行，也可以每小时运行。示例按 UTC 时间比较。如果付款日期要按客户所在时区计算，需要先确定统一的时区规则，再生成 next_billing_at。

from datetime import timedelta

from celery import shared_task
from django.db import transaction
from django.utils import timezone

from .models import BillingAction, Subscription


@shared_task
def dispatch_billing_actions():
    now = timezone.now()
    reminder_deadline = now + timedelta(days=10)

    reminder_candidates = Subscription.objects.filter(
        status=Subscription.Status.ACTIVE,
        next_billing_at__gt=now,
        next_billing_at__lte=reminder_deadline,
    )

    for subscription in reminder_candidates.iterator():
        action, created = BillingAction.objects.get_or_create(
            subscription=subscription,
            kind=BillingAction.Kind.SEND_PAYMENT_LINK,
            billing_at=subscription.next_billing_at,
            defaults={
                "idempotency_key": (
                    f"payment-link:{subscription.pk}:"
                    f"{subscription.next_billing_at.isoformat()}"
                ),
            },
        )
        if created:
            transaction.on_commit(
                lambda action_id=action.pk: execute_billing_action.delay(
                    action_id
                )
            )

    due_subscriptions = Subscription.objects.filter(
        status=Subscription.Status.ACTIVE,
        next_billing_at__lte=now,
    )

    for subscription in due_subscriptions.iterator():
        action, created = BillingAction.objects.get_or_create(
            subscription=subscription,
            kind=BillingAction.Kind.CHARGE,
            billing_at=subscription.next_billing_at,
            defaults={
                "idempotency_key": (
                    f"charge:{subscription.pk}:"
                    f"{subscription.next_billing_at.isoformat()}"
                ),
            },
        )
        if created:
            transaction.on_commit(
                lambda action_id=action.pk: execute_billing_action.delay(
                    action_id
                )
            )

如果支付服务商负责自动扣款，可以删除创建 CHARGE 操作的代码，只保留提醒、对账和 webhook 处理。

Celery Beat 可以按小时触发扫描：

from celery.schedules import crontab

CELERY_BEAT_SCHEDULE = {
    "dispatch-billing-actions-hourly": {
        "task": "billing.tasks.dispatch_billing_actions",
        "schedule": crontab(minute=0),
    },
}

扫描频率只决定任务最多会延迟多久。是否重复执行，应由数据库唯一约束和支付接口的 idempotency key 控制。

执行支付操作

具体如何调用支付 SDK，取决于所使用的支付服务商。以下代码只展示任务边界：

from celery import shared_task
from django.db import transaction

from .models import BillingAction


@shared_task(
    bind=True,
    autoretry_for=(TimeoutError, ConnectionError),
    retry_backoff=True,
    retry_jitter=True,
    max_retries=5,
)
def execute_billing_action(self, action_id):
    with transaction.atomic():
        action = (
            BillingAction.objects
            .select_for_update()
            .select_related("subscription")
            .get(pk=action_id)
        )

        if action.status == BillingAction.Status.SUCCEEDED:
            return

        action.status = BillingAction.Status.PROCESSING
        action.save(update_fields=["status"])

    try:
        if action.kind == BillingAction.Kind.SEND_PAYMENT_LINK:
            send_payment_link(
                subscription=action.subscription,
                idempotency_key=action.idempotency_key,
            )
        elif action.kind == BillingAction.Kind.CHARGE:
            charge_customer(
                subscription=action.subscription,
                idempotency_key=action.idempotency_key,
            )
        else:
            raise ValueError(f"Unsupported action kind: {action.kind}")
    except Exception as exc:
        BillingAction.objects.filter(pk=action_id).update(
            status=BillingAction.Status.FAILED,
            last_error=str(exc),
        )
        raise

    BillingAction.objects.filter(pk=action_id).update(
        status=BillingAction.Status.SUCCEEDED,
        last_error="",
    )

send_payment_link() 和 charge_customer() 是项目自己的支付适配层。调用支付 API 时，应把 BillingAction.idempotency_key 传给支持幂等请求的支付服务商。

生产环境还要考虑这种情况：worker 已经成功完成外部调用，却在写入 SUCCEEDED 前崩溃。支付请求可以使用 idempotency key 防止重复扣款。邮件或短信提醒则可以通过 outbox、服务商消息 ID 或可恢复的处理租约，减少重复发送。

每天运行一次是否合适

如果业务规则只是“付款日前 10 天发送”，没有具体的小时要求，每天运行一次通常就够了。查询时不要只匹配某个确切日期，而要扫描所有已经进入处理窗口且尚未成功处理的记录。这样即使任务停机一天，恢复后也能补发。

如果需要在用户购买时间对应的准确时刻执行，可以改为每 5 分钟或每小时扫描一次：

next_billing_at__lte=timezone.now()

数据库应为 next_billing_at 建立索引。数据量很大时，可以分批读取，并用数据库锁、状态字段或队列分片控制并发。

容易忽略的问题

不要用 timedelta(days=30) 表示“下个月”。每个月的天数不同，月末订阅也需要明确处理规则，例如 1 月 31 日的下一账期落在 2 月最后一天。
时间应统一使用带时区的 DateTimeField。开启 USE_TZ = True，并明确账单按 UTC、商户时区还是客户时区结算。
所有扣款操作都必须幂等。任务超时不代表支付失败，直接重试可能造成重复扣款。
webhook 同样需要幂等处理。可以保存支付平台的 event ID，并为其建立唯一约束。
Celery Beat 只负责投递或创建任务，实际扣款由 worker 执行。
不要保存原始银行卡号或 CVV。应使用支付服务商返回的 customer ID、payment method token 或 subscription ID。
订阅取消、暂停、付款方式失效、余额不足和退款，都需要分别处理相应状态。
本地账单状态与支付平台可能暂时不一致，因此需要安排周期性对账任务。

按天扫描数据库可以满足这类 scheduled payment 的需求。系统的可靠性取决于持久化账期、唯一约束、幂等调用、重试机制和 webhook 对账，无需为每个用户创建一个长期等待的定时任务。