PAYATHON 2026

订阅支付系统是否有更可靠的动态调度方案？

支付老李

结论

不建议为每个订阅单独创建或修改动态定时任务，也没必要因此引入 Apache Airflow。

更可靠的做法是：

  1. 以数据库中的 expires_at 作为判断订阅是否有效的唯一时间依据。
  2. 运行一个固定频率的 Celery 周期任务，批量扫描已过期订阅。
  3. 使用数据库索引、短事务和 SELECT ... FOR UPDATE SKIP LOCKED 减少锁竞争。
  4. 将通知、退款、资源回收等副作用拆成幂等的异步任务。
  5. 通过支付平台 webhook 更新订阅，并用周期性对账任务修复漏单。

采用这种设计后，即使 celery-beat 或 RabbitMQ 暂时延迟，用户也不会因为”过期任务尚未执行”而继续获得订阅权限。Airflow 更适合数据管道和存在依赖关系的批处理工作流，无法解决订阅状态建模、数据库锁和消息可靠性这些根本问题。

不要把定时任务当作订阅状态

如果系统必须等 expiration task 执行后才把用户标记为过期，那么任务延迟 60 秒，权限状态也会出错 60 秒。

订阅是否有效，应直接根据时间判断：

from django.utils import timezone


def has_active_subscription(subscription) -> bool:
    return (
        subscription.status == "active"
        and subscription.expires_at > timezone.now()
    )

如果业务中有宽限期，也应明确建模：

def has_subscription_access(subscription) -> bool:
    now = timezone.now()

    if subscription.status not in {"active", "past_due"}:
        return False

    access_until = subscription.grace_period_ends_at or subscription.expires_at
    return access_until > now

status 用来表达业务状态，expires_at 决定时间边界。系统不能依赖后台任务刚好在某一秒执行。

所有时间字段都应包含时区信息，并启用 Django 的时区支持：

USE_TZ = True
TIME_ZONE = "UTC"

向用户展示时间时，再转换到用户所在的时区。

用固定任务批量处理过期订阅

不要为每个用户创建一条 django_celery_beat.models.PeriodicTask。系统只需保留一个固定周期任务，例如每 10 秒或每分钟扫描一次。

# celery.py
from celery import Celery

app = Celery("project")
app.config_from_object("django.conf:settings", namespace="CELERY")
app.autodiscover_tasks()

app.conf.beat_schedule = {
    "expire-subscriptions": {
        "task": "subscriptions.tasks.expire_subscriptions",
        "schedule": 10.0,
        "options": {
            "queue": "subscriptions",
        },
    },
}

订阅表至少要为扫描条件建立索引：

from django.db import models


class Subscription(models.Model):
    class Status(models.TextChoices):
        ACTIVE = "active"
        PAST_DUE = "past_due"
        EXPIRED = "expired"
        CANCELED = "canceled"

    user = models.ForeignKey(
        "auth.User",
        on_delete=models.CASCADE,
        related_name="subscriptions",
    )
    status = models.CharField(
        max_length=20,
        choices=Status.choices,
        db_index=True,
    )
    expires_at = models.DateTimeField(db_index=True)
    expired_at = models.DateTimeField(null=True, blank=True)

    class Meta:
        indexes = [
            models.Index(
                fields=["status", "expires_at"],
                name="subscription_expiry_idx",
            ),
        ]

周期任务每次只处理有限数量的数据，以免单次事务持锁过久：

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

from .models import Subscription


BATCH_SIZE = 500


@shared_task(
    bind=True,
    autoretry_for=(Exception,),
    retry_backoff=True,
    retry_jitter=True,
    max_retries=5,
)
def expire_subscriptions(self):
    now = timezone.now()

    with transaction.atomic():
        subscriptions = list(
            Subscription.objects
            .select_for_update(skip_locked=True)
            .filter(
                status=Subscription.Status.ACTIVE,
                expires_at__lte=now,
            )
            .order_by("expires_at")[:BATCH_SIZE]
        )

        if not subscriptions:
            return {"expired": 0}

        subscription_ids = [item.id for item in subscriptions]

        Subscription.objects.filter(
            id__in=subscription_ids,
            status=Subscription.Status.ACTIVE,
            expires_at__lte=now,
        ).update(
            status=Subscription.Status.EXPIRED,
            expired_at=now,
        )

    dispatch_expiration_side_effects.delay(subscription_ids)
    return {"expired": len(subscription_ids)}

skip_locked=True 可用于 PostgreSQL 等支持该功能的数据库。不同数据库及版本的行为并不完全相同，使用前要确认当前数据库后端是否支持。

这种实现允许多个 worker 并行领取不同批次，无需互相等待。事务内只执行必要的状态变更，发送邮件、关闭外部资源等慢操作则放在事务外处理。

副作用必须幂等

Celery 通常采用”至少一次”语义。任务可能重试，也可能因 worker 在确认消息前退出而再次执行。因此，订阅过期后的操作不能假设自己只会执行一次。

from celery import shared_task

from .models import Subscription


@shared_task
def dispatch_expiration_side_effects(subscription_ids):
    subscriptions = Subscription.objects.filter(
        id__in=subscription_ids,
        status=Subscription.Status.EXPIRED,
    )

    for subscription in subscriptions.iterator():
        send_expiration_notification.delay(subscription.id)

通知任务可以通过唯一记录避免重复发送：

from django.db import transaction

from .models import Subscription, SubscriptionEvent


@shared_task
def send_expiration_notification(subscription_id):
    subscription = Subscription.objects.get(id=subscription_id)

    with transaction.atomic():
        event, created = SubscriptionEvent.objects.get_or_create(
            subscription=subscription,
            event_type="expiration_notification",
            defaults={"status": "pending"},
        )

        if not created:
            return

    # 调用邮件或消息服务。
    # 如果外部服务支持 idempotency key，应使用 event.id 作为幂等键。

对于”写数据库并发送消息”这类操作，数据库提交成功和消息发送之间仍然存在一个故障窗口。如果业务对可靠性要求较高，可以采用 transactional outbox：在同一个数据库事务中写入订阅状态和 outbox 事件，再由独立任务持续投递 outbox 中尚未发送的事件。

仅使用 transaction.on_commit() 可以防止事务回滚后仍然发送任务，但无法完全消除进程在提交后、发送前崩溃所造成的消息丢失。

支付状态应以 webhook 加对账为主

订阅续期不能只靠本地定时器推算。通常需要同时使用两条路径：

  • 支付平台 webhook：及时处理支付成功、扣款失败、退款和取消等事件。
  • 周期性对账：定期查询支付平台，修复 webhook 延迟、丢失或处理失败造成的数据不一致。

Webhook 也必须幂等。可以用支付平台提供的事件 ID 建立唯一约束：

class PaymentWebhookEvent(models.Model):
    provider = models.CharField(max_length=32)
    external_event_id = models.CharField(max_length=255)
    received_at = models.DateTimeField(auto_now_add=True)
    processed_at = models.DateTimeField(null=True, blank=True)

    class Meta:
        constraints = [
            models.UniqueConstraint(
                fields=["provider", "external_event_id"],
                name="unique_payment_webhook_event",
            ),
        ]

续期成功后，直接更新 expires_at。即使周期扫描任务已经读到旧数据，也要在更新时再次检查条件：

Subscription.objects.filter(
    id=subscription_id,
    status=Subscription.Status.ACTIVE,
    expires_at__lte=now,
).update(
    status=Subscription.Status.EXPIRED,
    expired_at=now,
)

如果 webhook 已经将 expires_at 延长到未来，这条更新不会误把订阅设为过期。

排查当前 60～150 秒延迟

先区分延迟发生在哪个环节：

  • 调度延迟：任务到期后，beat 过了很久才发送消息。
  • 队列延迟：消息已经进入 RabbitMQ，但 worker 很久以后才开始执行。

建议为任务记录三个时间点：

{
    "scheduled_at": "...",
    "published_at": "...",
    "started_at": "...",
}

然后计算：

scheduler_delay = published_at - scheduled_at
queue_delay     = started_at - published_at

如果是调度延迟，重点检查以下问题：

  • 是否只运行了一个 celery-beat 实例。
  • beat 是否与耗时服务运行在同一个进程或同一个资源受限的容器中。
  • django_celery_beat_periodictask 等表是否存在长事务或锁等待。
  • 系统是否频繁创建、修改大量 PeriodicTask。
  • 数据库连接是否耗尽，查询是否缺少索引。
  • 系统时钟、容器时钟和 Django 时区是否一致。

如果是队列延迟，则重点检查：

  • RabbitMQ 的队列堆积量和最老消息年龄。
  • worker 并发数是否不足。
  • 长任务是否与订阅任务共用队列。
  • worker 是否频繁重连、内存不足或进程阻塞。
  • 消息是否被限流、重试或重新入队。
  • RabbitMQ 节点是否触发内存或磁盘告警。

订阅状态任务最好使用独立队列：

celery -A project worker \
  --queues=subscriptions \
  --concurrency=4 \
  --loglevel=INFO

长耗时任务应放进其他队列，避免占满所有 worker：

CELERY_TASK_ROUTES = {
    "subscriptions.tasks.*": {"queue": "subscriptions"},
    "reports.tasks.*": {"queue": "reports"},
}

DatabaseScheduler 阻塞的常见原因

DatabaseScheduler 不适合承担高频、海量的”一用户一任务”调度。大量动态修改周期任务时，容易出现以下问题：

  • 更新调度表的事务持锁时间过长。
  • 应用事务尚未提交，scheduler 读取数据时需要等待锁。
  • 多个进程同时修改任务和调度变更标记。
  • 调度表不断增长，查询和同步成本随之上升。
  • 数据库连接池耗尽，调度进程无法及时查询。

如果确实需要修改调度记录，应缩小事务范围，并在业务事务提交后再执行调度操作：

from django.db import transaction


def update_subscription(subscription, new_expires_at):
    subscription.expires_at = new_expires_at
    subscription.save(update_fields=["expires_at"])

    transaction.on_commit(
        lambda: notify_subscription_changed.delay(subscription.id)
    )

不过，在订阅过期这个场景中，更合适的做法仍然是取消每个订阅对应的动态任务，只保留一条固定扫描任务。

是否需要替换 celery-beat

先改掉”一订阅一调度任务”的模型，通常比更换 scheduler 更重要。

如果改成固定扫描任务后仍有更高的可靠性要求，可以考虑：

  • 使用 Kubernetes CronJob、systemd timer 或平台提供的 cron，定期调用 Django management command。
  • 使用 PostgreSQL advisory lock，确保多个扫描进程中只有一个主扫描器运行。
  • 让多个扫描 worker 配合 SKIP LOCKED 并行处理。
  • 使用托管消息队列和独立监控，降低自行运维 RabbitMQ 的风险。
  • 对规模极大且需要精确定时的场景，使用专门的延迟队列或时间轮服务。

无论选择哪种调度器，都不能让调度器决定用户当前是否有权限。调度器只负责最终一致性的状态整理和副作用处理。

为什么不建议使用 Airflow

Airflow 擅长有向无环图、数据处理流水线、批处理依赖、任务重跑和历史管理。订阅系统需要处理的则是：

  • 低延迟的状态判断；
  • 高频业务事件；
  • webhook 幂等；
  • 数据库并发控制；
  • 消息重试与最终一致性。

引入 Airflow 会增加 metadata database、scheduler、executor、部署和监控等运维成本，但不会自动解决数据库锁、重复消息或过期权限判断错误。

更稳妥的架构如下：

支付 webhook
    │
    ▼
更新 Subscription.expires_at
    │
    ├── 请求时实时检查 expires_at
    │
    └── 固定周期扫描任务
            │
            ▼
       批量标记过期
            │
            ▼
       幂等副作用任务

周期对账任务 ───────► 修复支付状态不一致

对于小型和中型 Django 项目，这种设计更简单，也更容易测试、监控和恢复。系统不需要一个”绝不延迟”的调度器，而要保证在调度延迟、任务重复、消息丢失或进程重启时，订阅状态仍然正确。

备注：内容仅供参考。