celery定时任务简单使用

向右看齐 2023-06-21 15:26 74阅读 0赞

celery介绍  
Celery 是一个强大的分布式任务队列，它可以让任务的执行完全脱离主程序，甚至可以被分配到其他主机上运行。我们通常使用它来实现异步任务（ async task ）和定时任务（ crontab ）。 异步任务比如是发送邮件、或者文件上传, 图像处理等等一些比较耗时的操作 ，定时任务是需要在特定时间执行的任务。它的架构组成如下图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FpMjAwMGFp_size_16_color_FFFFFF_t_70][]

任务队列  
任务队列是一种跨线程、跨机器工作的一种机制.

任务队列中包含称作任务的工作单元。有专门的工作进程持续不断的监视任务队列，并从中获得新的任务并处理.

任务模块  
包含异步任务和定时任务。其中，异步任务通常在业务逻辑中被触发并发往任务队列，而定时任务由 Celery Beat 进程周期性地将任务发往任务队列。

消息中间件 Broker  
Broker ，即为任务调度队列，接收任务生产者发来的消息（即任务），将任务存入队列。 Celery 本身不提供队列服务，官方推荐使用 RabbitMQ 和 Redis 等。

任务执行单元 Worker  
Worker 是执行任务的处理单元，它实时监控消息队列，获取队列中调度的任务，并执行它。

任务结果存储 Backend  
Backend 用于存储任务的执行结果，以供查询。同消息中间件一样，存储也可使用 RabbitMQ, Redis 和 MongoDB 等。

使用 Celery 实现异步任务的步骤：  
(1) 创建一个 Celery 实例  
(2) 启动 Celery Worker ，通过delay() 或 apply\_async()(delay 方法封装了 apply\_async, apply\_async支持更多的参数 ) 将任务发布到broker  
(3) 应用程序调用异步任务

(4)存储结果 （发布的任务需要return才会有结果，否则为空）

Celery Beat：任务调度器，Beat进程会读取配置文件的内容，周期性地将配置中到期需要执行的任务发送给任务队列

使用 Celery 实现定时任务的步骤：  
(1) 创建一个 Celery 实例

(2) 配置文件中配置任务 ，发布任务 celery A xxx beat

(3) 启动 Celery Worker

(4) 存储结果

celery定时任务简单使用  
以下是使用celery实现一个定时任务的demo，能够良好的定时执行。

目录结构如下

shylin@shylin:~/Desktop$ tree celery_task
    celery_task
    ├── celeryconfig.py    # celeryconfig配置文件
    ├── celeryconfig.pyc
    ├── celery.py   # celery对象
    ├── celery.pyc
    ├── epp_scripts   # 任务函数
    │   ├── __init__.py
    │   ├── __init__.pyc
    │   ├── test1.py
    │   ├── test1.pyc
    │   ├── test2.py
    │   └── test2.pyc
    ├── __init__.py
    └── __init__.pyc

celery配置文件 celeryconfig.py

from __future__ import absolute_import # 拒绝隐式引入，因为celery.py的名字和celery的包名冲突，需要使用这条语句让程序正确地运行
    from celery.schedules import crontab
    
    broker_url = "redis://127.0.0.1:6379/5"  
    result_backend = "redis://127.0.0.1:6379/6"
    
    broker_url = "redis://127.0.0.1:6379/2"   # 使用redis存储任务队列
    result_backend = "redis://127.0.0.1:6379/6"  # 使用redis存储结果
    
    task_serializer = 'json'
    result_serializer = 'json'
    accept_content = ['json']
    timezone = "Asia/Shanghai"  # 时区设置
    worker_hijack_root_logger = False  # celery默认开启自己的日志，可关闭自定义日志，不关闭自定义日志输出为空
    result_expires = 60 * 60 * 24  # 存储结果过期时间（默认1天）
    
    # 导入任务所在文件
    imports = [
        "celery_task.epp_scripts.test1",  # 导入py文件
        "celery_task.epp_scripts.test2",
    ]
    
    
    # 需要执行任务的配置
    beat_schedule = {
        "test1": {
            "task": "celery_task.epp_scripts.test1.celery_run",  #执行的函数
            "schedule": crontab(minute="*/1"),   # every minute 每分钟执行 
            "args": ()  # # 任务函数参数
        },
    
        "test2": {
            "task": "celery_task.epp_scripts.test2.celery_run",
            "schedule": crontab(minute=0, hour="*/1"),   # every minute 每小时执行
            "args": ()
        },
    
    }
    
    "schedule": crontab（）与crontab的语法基本一致
    "schedule": crontab(minute="*/10",  # 每十分钟执行
    "schedule": crontab(minute="*/1"),   # 每分钟执行
    "schedule": crontab(minute=0, hour="*/1"),    # 每小时执行

celery初始化文件

# coding:utf-8
    from __future__ import absolute_import # 拒绝隐式引入，因为celery.py的名字和celery的包名冲突，需要使用这条语句让程序正确地运行
    from celery import Celery
    
    # 创建celery应用对象
    app = Celery("celery_demo")
    
    # 导入celery的配置信息
    app.config_from_object("celery_task.celeryconfig")

任务函数（epp\_scripts目录下）

# test1.py
    from celery_task.celery import app
    
    def test11():
        print("test11----------------")
    
    def test22():
        print("test22--------------")
        test11()
    
    @app.task
    def celery_run():
        test11()
        test22()
    
    if __name__ == '__main__':
        celery_run()
    ------------------------------------------------------------
    # test2.py
    from celery_task.celery import app
    
    def test33():
        print("test33----------------")
        # print("------"*50)
    
    def test44():
        print("test44--------------")
        # print("------" * 50)
        test33()
    
    @app.task
    def celery_run():
        test33()
        test44()
    
    
    if __name__ == '__main__':
        celery_run()

发布任务

# 在celery_task同级目录下执行
    shylin@shylin:~/Desktop$ celery -A celery_task beat
    celery beat v4.2.0 (windowlicker) is starting.
    __    -    ... __   -        _
    LocalTime -> 2018-06-29 09:42:02
    Configuration ->
        . broker -> redis://127.0.0.1:6379/5
        . loader -> celery.loaders.app.AppLoader
        . scheduler -> celery.beat.PersistentScheduler
        . db -> celerybeat-schedule
        . logfile -> [stderr]@%WARNING
        . maxinterval -> 5.00 minutes (300s)

执行任务

# 在celery_task同级目录下执行
    shylin@shylin:~/Desktop$ celery -A celery_task worker --loglevel=info
    
     -------------- celery@shylin v4.2.0 (windowlicker)
    ---- **** ----- 
    --- * ***  * -- Linux-4.15.0-23-generic-x86_64-with-Ubuntu-18.04-bionic 2018-06-29 12:06:53
    -- * - **** --- 
    - ** ---------- [config]
    - ** ---------- .> app:         belletone:0x7f5b876f1a10
    - ** ---------- .> transport:   redis://127.0.0.1:6379/5
    - ** ---------- .> results:     redis://127.0.0.1:6379/6
    - *** --- * --- .> concurrency: 8 (prefork)
    -- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
    --- ***** ----- 
     -------------- [queues]
                    .> celery           exchange=celery(direct) key=celery
    
    
    [tasks]
      . celery_task.epp_scripts.test1.celery_run
      . celery_task.epp_scripts.test2.celery_run
    
    [2018-06-29 12:06:54,107: INFO/MainProcess] Connected to redis://127.0.0.1:6379/5
    [2018-06-29 12:06:54,116: INFO/MainProcess] mingle: searching for neighbors
    [2018-06-29 12:06:55,143: INFO/MainProcess] mingle: all alone
    [2018-06-29 12:06:55,161: INFO/MainProcess] celery@shylin ready.
    [2018-06-29 12:07:00,073: INFO/MainProcess] Received task: celery_task.epp_scripts.test2.celery_run[f4522425-b744-4f1a-8c6c-eb37ab99842b]  
    [2018-06-29 12:07:00,075: INFO/MainProcess] Received task: celery_task.epp_scripts.test1.celery_run[3e00aa9c-0947-49b9-8ee4-cc75d6dc37ab]  
    [2018-06-29 12:07:00,078: WARNING/ForkPoolWorker-6] test33----------------
    [2018-06-29 12:07:00,079: WARNING/ForkPoolWorker-6] test44--------------
    [2018-06-29 12:07:00,079: WARNING/ForkPoolWorker-6] test33----------------
    [2018-06-29 12:07:00,079: WARNING/ForkPoolWorker-4] test11----------------
    [2018-06-29 12:07:00,081: WARNING/ForkPoolWorker-4] test22--------------
    [2018-06-29 12:07:00,081: WARNING/ForkPoolWorker-4] test11----------------
    [2018-06-29 12:07:00,094: INFO/ForkPoolWorker-6] Task celery_task.epp_scripts.test2.celery_run[f4522425-b744-4f1a-8c6c-eb37ab99842b] succeeded in 0.0169868329995s: None
    [2018-06-29 12:07:00,094: INFO/ForkPoolWorker-4] Task celery_task.epp_scripts.test1.celery_run[3e00aa9c-0947-49b9-8ee4-cc75d6dc37ab] succeeded in 0.0161407030009s: None

celery相关命令

# 在celery_task同级目录下执行   celery worker/beat xxx
    celery -A celery_task beat  # 发布任务
    celery -A celery_task worker --loglevel=info  # 执行任务
    celery -B -A celery_task worker --loglevel=info  # 合并成一条
    
    /home/shylin/.virtualenvs/belle/bin/celery -B -A /home/shylin/Desktop/sky_server worker --loglevel=info
    
    
    command= /usr/local/thirdparty/sky_server_env/bin/celery  -B -A celery_task worker
    directory=/usr/local/cloud   # celery_task work不确定是否可行？
    # 注意修改broker路径
    # celery_task放在 /usr/local/cloud/
    python -m celeryconfig # 检查配置文件
    
    nohup /usr/local/thirdparty/sky_server_env/bin/celery  -B -A celery_task worker -l info --workdir=/usr/local/cloud/ &     # 启动命令

定时方式

from celery.schedules import crontab
    from datetime import timedelta
    ......
    
    方式一：
     "schedule": timedelta(seconds=30), # hours=xx,minutes=xx 每小时/每分钟  （此项可以精确到秒）
    
    方式二：
     "schedule": crontab(minute="*/10"),   # every 10 minutes

# 后台启动 celery worker进程 
    celery multi start work_1 -A appcelery  
    # work_1 为woker的名称，可以用来进行对该进程进行管理
    
    # 多进程相关
    celery multi stop WOERNAME      # 停止worker进程,有的时候这样无法停止进程，就需要加上-A 项目名，才可以删掉
    celery multi restart WORKNAME        # 重启worker进程
    
    # 查看进程数
    celery status -A celery_task       # 查看该项目运行的进程数   celery_task同级目录下
    
    执行完毕后会在当前目录下产生一个二进制文件，celerybeat-schedule 。
    该文件用于存放上次执行结果：
    　　1、如果存在celerybeat-schedule文件，那么读取后根据上一次执行的时间，继续执行。
    　　2、如果不存在celerybeat-schedule文件，那么会立即执行一次。
    　　3、如果存在celerybeat-schedule文件，读取后，发现间隔时间已过，那么会立即执行。

————————————————  
版权声明：本文为CSDN博主「Shyllin」的原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接及本声明。  
原文链接：https://blog.csdn.net/Shyllin/article/details/80940643

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FpMjAwMGFp_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191213143816466.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FpMjAwMGFp,size_16,color_FFFFFF,t_70