如何在 Django 中创建一个简单的调度程序

百变鹏仔 4天前 #Python

文章标签创建一个

如果您需要每 x 分钟/秒等运行一个函数来进行一些清理，触发一些操作，您可以在线程模块和 django 自定义 cli 命令的帮助下执行一个简单的调度程序。

假设我想每 5 秒调用一个函数以在外部 api 上发布一些内容。

在您的 django 应用程序中创建一个名为 management 的文件夹/包，在该文件夹内创建另一个名为commands 的文件夹。在命令文件夹中创建一个名为 runposter.py 的模块。最后你会得到类似这样的结构 yourapp/management/commands/runposter.py。

在此代码中，我们使用一个运行 while 循环的线程，只要它不每 5 秒停止一次。将 print("posting") 替换为您要运行的函数/逻辑。

# runposter.pyimport timefrom threading import Thread, Eventfrom django.conf import settingsfrom django.core.management.base import BaseCommandstop_event = Event()def my_job():    while not stop_event.is_set():        try:            print("posting")            time.sleep(5)        except KeyboardInterrupt:            breakclass Command(BaseCommand):    help = "Run Poster."    def handle(self, *args, **options):        poster = Thread(target=my_job)        try:            print("Starting poster...")            poster.start()            while poster.is_alive():                poster.join(timeout=1)        except KeyboardInterrupt:            print("Stopping poster...")            stop_event.set()            poster.join()            print("Poster shut down successfully!")

很好，现在打开另一个终端窗口并运行 python manage.py runposter。如您所见，命令 runposter 是根据我们给出的模块名称创建的。

当然，对于更复杂的事情，我建议使用 rq-scheduler 或 celery 周期性任务。

但是，对于简单的情况，这应该足够了。

文章推荐

如何在 Django 中创建一个简单的调度程序

如何用Python编写求解最小公倍数的算法？

如何使用Python实现贪心算法？

如何使用Python实现基数排序算法？

如何使用Python实现回归分析算法？

如何用Python编写深度优先搜索算法？