Python计算当前与上次登录时间差 优化Celery自动通信提醒任务逻辑

实现方案 #

核心优化思路 #

• 用策略模式封装所有通知规则，完全替代if分支，后续新增规则只需新增配置项，无需修改核心逻辑
• 优化数据库查询逻辑，减少全表扫描和循环内冗余操作，降低数据库负载
• 拆分Celery任务职责，主任务仅做筛选分发，实际发信逻辑走异步子任务，提升执行效率

具体实现步骤 #

1. 预定义通知规则配置

把所有触发规则统一维护为配置列表，每条规则包含触发条件、关联模板、通知类型等核心属性：

from datetime import timedelta
from django.conf import settings
# 通知规则配置，新增规则只需往该列表加配置项即可
NOTIFICATION_RULES = [
    {
        "trigger_delay": timedelta(hours=2),
        "template_key": "automated_comms_signup_uncompleted",
        "notify_type": NotificationTypes.EMAIL_NOTIFICATION,
        "notification_desc": "注册未完成2小时提醒"
    },
    # 后续新增其他规则直接加在这里即可，示例：7天未登录提醒
    # {
    #     "trigger_delay": timedelta(days=7),
    #     "template_key": "automated_comms_7d_inactive",
    #     "notify_type": NotificationTypes.PUSH_NOTIFICATION,
    #     "notification_desc": "7天未登录提醒"
    # }
]

2. 改造Celery任务逻辑

首先优化查询逻辑，避免全表拉取已通知用户，同时直接在SQL层筛选符合时间条件的用户，减少内存占用，最后写入已通知记录避免重复发送：

@app.task
def check_registered_users():
    from apps.users.models import User
    from apps.notifications.models import AutomatedCommunicationNotifications
    from apps.system.models import SiteConfiguration
    from datetime import datetime, timedelta

    now = datetime.utcnow()
    # 全局配置只查询一次，避免循环内重复请求数据库
    site_config = SiteConfiguration.get_solo()
    # 拉取已发送记录，存为集合方便快速判断
    sent_notifications = AutomatedCommunicationNotifications.objects.values_list("user_id", "type")
    sent_set = set(sent_notifications)

    # 遍历规则逐个处理，无需修改核心逻辑即可扩展规则
    for rule in NOTIFICATION_RULES:
        # 计算当前规则的触发时间阈值
        trigger_time_threshold = now - rule["trigger_delay"]
        # SQL层直接过滤符合条件的用户，避免拉取全量用户后循环判断
        eligible_users = User.objects.filter(
            is_active=True,
            last_login__lte=trigger_time_threshold,
            date_joined__gte=now - timedelta(days=30) # 限制查询范围，避免拉取过久的历史用户
        ).exclude(id__in=[uid for uid, ntype in sent_set if ntype == rule["notify_type"]])

        # 批量写入用的缓存列表，进一步降低数据库写入压力
        create_buffer = []
        for user in eligible_users:
            # 调用异步发信任务，避免主任务阻塞
            template = getattr(site_config, rule["template_key"])
            send_plain_email_task.delay(
                email=user.email,
                subject=template.subject,
                html_message=template.content,
                from_email=f'{settings.EMAIL_FROM_PREFIX} <{settings.DEFAULT_FROM_EMAIL}>',
            )
            # 先存入缓存列表，后续批量写入
            create_buffer.append(AutomatedCommunicationNotifications(
                user=user,
                type=rule["notify_type"]
            ))
        # 批量写入已发送记录
        if create_buffer:
            AutomatedCommunicationNotifications.objects.bulk_create(create_buffer)

3. 可选性能优化

如果用户量较大，可以做如下优化进一步降低系统负载：

• 给User表的last_login、date_joined字段加索引，大幅提升过滤查询速度
• 把用户筛选逻辑按注册时间分片，拆分为多个子异步任务并行执行，避免单任务查询数据量过大
• 定期清理AutomatedCommunicationNotifications表中超过30天的历史记录，降低表体积
• 发信任务增加幂等校验，避免极端情况重复发送通知

注意事项 #

• 定期任务的执行频率要匹配最小的触发时间间隔，比如最小触发间隔是1小时，就设置每小时执行一次即可
• 模板配置可以抽成独立的模型维护，不用绑定在SiteConfiguration里，后续修改模板不需要调整代码

内容的提问来源于stack exchange，提问作者user9745220