利用Celery实现定时任务

末蓝、 2023-02-28 01:56 46阅读 0赞

`Celery`是一个基于分布式消息传输的异步任务队列，它专注于实时处理，同时也支持任务调度。在之前的文章[Python之celery的简介与使用][Python_celery]中，笔者简单介绍了celery以及celery的使用。本文将会将会如何利用celery来实现定时任务。  
  定时任务，在我们平时的工作中并不少见，比如定时备份数据库，清理日志，以及对数据定时做统计等。  
  本文将会结合两个定时任务来进行讲解，这两个定时任务为：

*  每10秒做加法
 *  每1分钟告诉时间

整个项目的演示代码很简单，如下：  
![项目代码][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2pjbGlhbjkx_size_16_color_FFFFFF_t_70_pic_center]  
  `celeryconfig.py`文件为celery的相关配置，代码如下：

BROKER_URL = 'redis://localhost'    # 使用Redis作为消息代理
    
    CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'  # 把任务结果存在了Redis
    
    CELERY_TASK_SERIALIZER = 'msgpack'  # 任务序列化和反序列化使用msgpack方案
    
    CELERY_RESULT_SERIALIZER = 'json'   # 读取任务结果一般性能要求不高，所以使用了可读性更好的JSON
    
    CELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24   # 任务过期时间，不建议直接写86400，应该让这样的magic数字表述更明显
    
    CELERY_ACCEPT_CONTENT = ['json', 'msgpack']     # 指定接受的内容类型
    
    CELERYD_CONCURRENCY = 10    # 并发worker数

`tasks.py` 为具体的任务情况，这里只配置了上述两个任务，代码如下：

from proj.app_test import app
    import datetime
    from random import randint
    
    
    @app.task
    def add(x, y):
        a = randint(x, y)
        b = randint(x, y)
        return "%s + %s = %s" % (a, b, a+b)
    
    
    @app.task
    def time_teller():
        return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")

`app_test.py`中是程序的运行配置以及定时任务的设置，代码如下：

from celery import Celery
    from celery.schedules import crontab
    
    app = Celery('proj', include=['proj.tasks'])
    app.config_from_object('proj.celeryconfig')
    
    # 定时任务
    app.conf.beat_schedule = { 
        "each10s_task": { 
            "task": "proj.tasks.add",
            "schedule": 10,
            "args": (0, 100)
        },
        "each1m_task": { 
            "task": "proj.tasks.time_teller",
            "schedule": crontab(minute="*/1")
        }
    }

从上述的代码中，我们可以看出，celery的定时任务实现方式应该借助了Linux中Crontab的实现机制。在具体的配置中，需要指定任务的名称（比如：`each10s_task`），任务的实现函数（比如：`proj.tasks.add`）以及定时时间和函数参数（非必须）。关于定时的schedule模式，可以参考网址：[https://docs.celeryproject.org/en/stable/userguide/periodic-tasks.html][https_docs.celeryproject.org_en_stable_userguide_periodic-tasks.html] 。  
  接下来，我们来模拟该定时任务的运行。  
  首先，需要启动消息队列，这里选择`Redis`。接着，运行任务调度的命令，如下：

celery beat -A proj.app_test

输出如下：  
![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2pjbGlhbjkx_size_16_color_FFFFFF_t_70_pic_center 1][]  
接着运行celery的定时任务，输出结果如下：  
![运行结果][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2pjbGlhbjkx_size_16_color_FFFFFF_t_70_pic_center 2]  
  如果我们想在网页中可视化地查看celery的实时监控情况，可以再输入命令（事先安装`flower`）：

celery -A proj.app_test flower

在网页中输入localhost:5555，可以看到worker的运行情况，如下：  
![flower监控页面][flower]  
  本文简单讲述了如何利用celery来实现定时任务，其实有很多实现定时任务的办法，比如crontab等等。笔者主要是为了研究celery，后续还会再写这方面的文章，欢迎大家关注~

[Python_celery]: https://blog.csdn.net/jclian91/article/details/86749032
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2pjbGlhbjkx_size_16_color_FFFFFF_t_70_pic_center]: /images/20230209/891a4a3a3d4b40f4b5dd7e4c2dfc44f8.png
[https_docs.celeryproject.org_en_stable_userguide_periodic-tasks.html]: https://docs.celeryproject.org/en/stable/userguide/periodic-tasks.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2pjbGlhbjkx_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20230209/a5cf9a1fcaef462a95a104ac25f07eea.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2pjbGlhbjkx_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20230209/0e049885ae8d4e3ab980ff053dc00e11.png
[flower]: /images/20230209/ee0d380417544975bceded261089d922.png