安排重复事件的提醒

Posted 2023-02-22

【中文标题】安排重复事件的提醒

【英文标题】：Schedule reminder for recurring event

【发布时间】：2016-07-16 10:12:15

【问题描述】：

我正在使用一个网络应用程序，它允许用户在日历上创建事件（一次性或重复），并且在事件开始前不久，系统会通知其参与者。我在设计此类通知的流程时遇到了麻烦，尤其是对于重复性事件。

需要考虑的事项：

Web 应用程序的体系结构使得有许多相同结构的数据库，每个数据库都有自己的一组用户和事件。因此，针对一个数据库的任何查询都需要针对数千个其他数据库进行。

重复事件可能包含排除日期（类似于 RRULE 和 EXDATE 组合）。

用户可以更新事件的时间/循环规则。

该应用程序是用 Python 编写的，并且已经使用带有 Redis 代理的 Celery 3.1。使用此设置的解决方案会很好，尽管任何事情都可以。根据我的发现，目前很难用 Celery 动态添加周期性任务。

我正在尝试的解决方案：

定期任务每天运行一次，扫描每个数据库并添加任务以在适当的时间针对当天重复发生的每个事件进行通知。

如上生成的每个任务都将其 id 临时保存在 Redis 中。如果用户在其通知任务安排后更改了当天的事件时间，该任务将被撤销并替换为新的。

上述解决方案的示例代码：

在tasks.py，所有要运行的任务：

from celery.task import task as celery_task
from celery.result import AsyncResult
from datetime import datetime

# ...

@celery_task
def create_notify_task():
    for account in system.query(Account):
        db_session = account.get_session()    # get sql alchemy session
        for event in db_session.query(Event):
            schedule_notify_event(account, partial_event)


@celery_task(name='notify_event_users')
def notify_event_users(account_id, event_id):
    # do notification for every event participant
    pass

def schedule_notify_event(account, event):
    partial_event = event.get_partial_on(datetime.today())
    if partial_event:
        result = notify_event_users.apply_async(
                args = (account.id, event.id),
                eta = partial_event.start)
        replace_task_id(account.id, event.id, result.id)
    else:
        replace_task_id(account.id, event.id, None)

def replace_task_id(account_id, event_id, result_id):
    key = ':event'.format(account_id)
    client = redis.get_client()
    old_result_id = client.hget(key, event_id)
    if old_result_id:
        AsyncResult(old_result_id).revoke()
    client.hset(key, event_id, result_id)

在event.py:

# when a user change event's time
def update_event(event, data):
    # ...
    # update event
    # ...
    schedule_notify_event(account, event)

Celery 设置文件：

from celery.schedules import crontab

CELERYBEAT_SCHEDULE = 
    'create-notify-every-day': 
        'task': 'tasks.create_notify_task',
        'schedule': crontab(minute=0, hour=0),
        'args': (,)
    ,

上述的一些缺点是：

每日任务可能需要很长时间才能运行。最后处理的数据库中的事件必须等待并且可能会丢失。提前安排该任务（例如第二天前 2 小时）可能会缓解这种情况，但是首次运行设置（或在服务器重新启动后）有点尴尬。

必须注意不要为同一事件安排两次通知任务（例如，因为 create_notify_task 每天运行多次...）。

有没有更明智的方法来解决这个问题？

相关问题：

Efficient recurring tasks in celery?

【问题讨论】：

您需要发布您已经尝试编写的任何代码。这不是我们为您工作的承包商网站。您必须进行第一次尝试，我们将在您进行过程中帮助您解决问题。

我只要求一种方法/工作流程，而不是任何代码；我已经提出了一个我想到的粗略的解决方案。无论如何，我已经添加了一些代码 sn-ps 给你一个更清晰的画面。

我不是 Python 大佬，所以希望其他 Python 专家可以跳到这里，如果他们确实看到问题，可以做出回应。但是从你所展示的内容来看，我真的没有看到你的逻辑有什么问题。

好吧，至少我认为它有效。但正如我所概述的那样，有一些缺点，这让它感觉很脆弱。该方法实际上不必是特定于 Python 的，尽管我会回避类似于构建自己的任务调度程序的东西。

【参考方案1】：

很久没有答案了，我忘记了这个问题。无论如何，当时我采用了以下解决方案。我在这里概述一下，以防有人感兴趣。

创建事件后，任务计划在其下一次发生（即下一次通知时间）之前不久运行。计划时间是在应用所有重复规则和例外规则的情况下计算得出的，因此它只是一个简单的 celery 计划一次性任务。 当任务运行时，它会执行通知作业，并在下一个通知时间安排新任务（同样，考虑所有重复规则和例外规则）。如果没有下一个事件发生，则不会安排新任务。 任务 ID 与事件一起保存在数据库中。如果更改事件的时间，则取消任务并在新的下一个通知时间安排新任务。当任务运行并安排新任务时，新任务的 id 会保存在数据库中。

我能想到的一些优点和缺点：

优点： 在 celery 中不需要复杂的重复规则，因为任务只安排一次运行。 每个任务都相当小而且很快，因为它只需要关心一个事件通知。 缺点： 任何时候都有大量的celery定时任务等待执行，大概有几十万的量级。我不确定这如何影响芹菜的性能，所以它可能是也可能不是真正的骗局。到目前为止，系统似乎运行良好。

【讨论】：

还在用 Redis 作为后端吗？我们有一个使用等效设置的系统，但如果作业被安排到未来很长一段时间，它就不能很好地工作。如果作业的安排超出 VISIBILITY_TIMEOUT 设置，则作业可能会安排多次。你遇到过这个问题吗？

是的，仍然是 Redis 后端。在这部分投入生产之前，我已经转移到另一个项目，所以我没有听到更多关于它的信息。这似乎是 Redis 传输的一个已知且长期存在的问题（例如 here...），甚至是 documented

没错，所以我想知道如果这对您来说是个问题，您是如何解决的。

好吧，我是在看了你的评论后才知道这个问题的。在我参与该项目期间，从未发现过此类问题。至于解决方案，人们讨论了一些变通方法，例如使用另一个代理（例如 AMQP）或增加 visibility_timeout。他们对你的案子有用吗？我刚才想到的另一个粗略的解决方法是仅在短时间内安排任务（小于 visibility_timeout），然后任务必须在短时间内继续重新安排自己，直到达到预期的实际 ETA，此时它可以执行工作并完成（不再重新安排）。