如何在Heroku中使用Django安排一次性事件？

Question

我有一个关于在Heroku的分布式环境中调度未来事件所需的软件设计的问题。

我认为写明我想要实现什么是更好的做法，但我肯定已经进行了研究，即使工作了两个小时，我也无法自己解决它。

假设在我的views.py中，我有一个函数：

def after_6_hours():
    print('6 hours passed.')


def create_game():
    print('Game created')
    # of course time will be error, but that's just an example
    scheduler.do(after_6_hours, time=now + 6)

我想要实现的目标是在create_game被调用后确切地6小时运行after_6_hours函数。如您所见，此功能定义于clock.py或task.py等普通文件之外。

那么，我该如何让整个应用程序在Heroku上始终运行，并能够将此作业添加到虚构的scheduler库的队列中呢？

值得一提的是，我不能使用Heroku的Temporizer插件。 APScheduler和Python rq的组合看起来很有前途，但是示例很简单，都在clock.py的同一文件中进行调度，而我不知道如何将所有内容与我的设置绑定在一起。谢谢！

- Sam Hosseini

---

所以如果我理解正确的话，您有一个分布在多个节点上的Web应用程序，并且想要在其中一个节点上6小时后运行一个任务（调用函数）...为什么不使用Celery呢？ - Peter Brittain

@PeterBrittain Heroku已经将节点及其环境抽象化了，称之为dyno。我的关注点不在节点上，而是希望能够在这个环境中拥有一个完全功能的作业调度系统，但我无法想到一种方法。我所考虑的是可能需要3个dynos，一个用于服务我的应用程序，一个用于调度器（它将解析预定的作业），还有一个用于执行这些任务。在我的“views.py”中，我希望能够添加到预定作业的数据库中，以便稍后调度器dyno可以解析它。现在，我想准备好这个设置。 - Sam Hosseini

我可以通过适当的文档来启动Celery或基本上任何其他作业调度程序。细节并不重要。我正在寻找拥有这个任务/作业调度系统的更大图景。 - Sam Hosseini

Answer 1

在Heroku上，您可以在Web Dyno中运行Django应用程序，它将负责提供应用程序并安排任务。例如（请注意，我没有测试运行代码）：

创建after_hours.py，其中包含您要安排的函数（请注意，在worker中我们也将使用相同的源代码）。

def after_6_hours():
        print('6 hours passed.')

在您的 views.py 中使用 rq（请注意，在您的情况下，仅使用 rq 是不够的，因为您需要安排任务），以及 rq-scheduler：

from redis import Redis
from rq_scheduler import Scheduler
from datetime import timedelta

from after_hours import after_6_hours

def create_game():
    print('Game created')

    scheduler = Scheduler(connection=Redis()) # Get a scheduler for the "default" queue
    scheduler.enqueue_in(timedelta(hours=6), after_6_hours) #schedules the job to run 6 hours later.

调用create_game()函数应该会安排6个小时后运行after_6_hours()函数。
提示: 您可以使用Redis To Go插件在Heroku中提供Redis。

下一步是运行rqscheduler工具，它每分钟轮询Redis以查看是否有在那个时间执行的任务，并将其放入队列（rq工人将监听该队列）。

现在，在Worker Dyno上创建一个名为after_hours.py的文件。

def after_6_hours():
    print('6 hours passed.')
    #Better return something

并创建另一个文件worker.py：

import os

import redis
from rq import Worker, Queue, Connection

from after_hours import after_6_hours

listen = ['high', 'default', 'low'] # while scheduling the task in views.py we sent it to default

redis_url = os.getenv('REDISTOGO_URL', 'redis://localhost:6379')

conn = redis.from_url(redis_url)

if __name__ == '__main__':
    with Connection(conn):
        worker = Worker(map(Queue, listen))
        worker.work()

并运行此 worker.py

python worker.py

那应该在Worker Dyno中运行预定任务（在这种情况下是afer_6_hours）。请注意，关键是也要将相同的源代码（在这种情况下是after_hours.py）提供给worker。在rqdocs中也强调了这一点。

确保worker和工作生成器完全共享相同的源代码。

如果需要，docs中有一个提示来处理不同的代码库。

For cases where the web process doesn't have access to the source code running in the worker (i.e. code base X invokes a delayed function from code base Y), you can pass the function as a string reference, too.

 q = Queue('low', connection=redis_conn)
 q.enqueue('my_package.my_module.my_func', 3, 4)

希望rq-scheduler也能尊重这种传递字符串而不是函数对象的方式。

只要你理解这个问题，就可以使用任何模块/调度工具（Celery/RabbitMQ，APScheduler等）。

- LearnerEarner