异步任务神器 Celery 快速入门

柔光的暖阳◎ 2021-09-13 23:48 562阅读 0赞

# 简介 #

在程序运行过程中，要执行一个很久的任务，但是我们又不想主程序被阻塞，常见的方法是多线程。可是当并发量过大时，多线程也会扛不住，必须要用线程池来限制并发个数，而且多线程对共享资源的使用也是很麻烦的事情。还有协程，但是协程毕竟还是在同一线程内执行的，如果一个任务本身就要执行很长时间，而不是因为等待IO被挂起，那其他协程照样无法得到运行。

在程序的运行过程中，我们经常会碰到一些耗时耗资源的操作，为了避免它们阻塞主程序的运行，我们经常会采用多线程或异步任务。比如，在 Web 开发中，对新用户的注册，我们通常会给他发一封激活邮件，而发邮件是个 IO 阻塞式任务，如果直接把它放到应用当中，就需要等邮件发出去之后才能进行下一步操作，此时用户只能等待再等待。更好的方式是在业务逻辑中触发一个发邮件的异步任务，而主程序可以继续往下运行。

对于以上两种问题的解决的办法，本文要介绍一个强大的分布式任务队列Celery，它可以让任务的执行同主程序完全脱离，甚至可以被分配到其他主机上运行。它通过队列来调度任务，不用担心并发量高时系统负载过大。它可以用来处理复杂系统性能问题，却又相当灵活易用。我们通常使用它来实现异步任务（async task）和定时任务（crontab）。下面我们就来了解下Celery。

--------------------

# Celery的架构组成 #

**Celery的架构组成图**

![这里写图片描述][SouthEast]

**或者**

![这里写图片描述][SouthEast 1]

由上图可以看到，Celery 主要包含以下几个模块：

## 任务模块 Task ##

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

## 消息中间件 Broker ##

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

## 任务执行单元 Worker ##

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

## 任务结果存储 Backend ##

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

--------------------

# Celery 准备工作 #

假设已购买4台阿里云服务器，4台机器环境配置都是相同，以其中一台机器为例：

首先使用 ssh 登录 Ubuntu 云端，开始进行如下配置：

**Redis-server 的安装：**

apt install redis-server

**Celery的安装：**

apt update
    apt install python3-pip
    pip3 install celery

**redis的安装：**

pip3 install redis

其它的模块，诸如 requests模块，bs4模块等，根据实际项目需要，可以进行安装。

至此一台云端配置完毕，其余三台按此配置即可。

下面的简单实例中，可将 **celery\_demo**文件中的 **celery\_app** 文件夹，放在云端的 **/home/** 中。

**在 celery\_demo 文件中，可以通过如下命令操作：**

scp -r celery_app root@xxx.xxx.111.168:/home/celery_app

--------------------

# Celery 快速入门的简单示例 #

为了简单起见，对于 Broker 和 Backend，这里都使用 redis，并开启 redis 服务。

首先，在开始建立如下两个Celery项目之前，推荐的做法是将配置项统一写入到一个配置文件中，通常我们将该文件命名为 celeryconfig.py。

## 异步任务 ##

**使用 Celery 实现异步任务主要包含三个步骤：**

1.  创建一个 Celery 实例
2.  启动 Celery Worker
3.  应用程序调用异步任务

Celery 的异步任务配置比较多，可以在[**官方文档**][Link 1]查询每个配置项的含义。

**建立的项目结构如下：**

celery_demo                    # 项目根目录
        ├── celery_app             # 存放 celery 相关文件
        │   ├── __init__.py
        │   ├── celeryconfig.py    # 配置文件
        │   ├── task1.py           # 任务文件 1
        │   └── task2.py           # 任务文件 2
        └── client.py              # 应用程序

***\_init\_*.py 代码如下：**

# -*- coding: utf-8 -*-
    
    from celery import Celery
    
    app = Celery('demo')                                # 创建 Celery 实例
    app.config_from_object('celery_app.celeryconfig')   # 通过 Celery 实例加载配置模块

**celeryconfig.py 代码如下：**

BROKER_URL = 'redis://xxx.xxx.123.168:6379'         # 指定 Broker 即服务器或者本地的地址
    CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/0'  # 指定 Backend
    
    CELERY_TIMEZONE='Asia/Shanghai'                     # 指定时区，默认是 UTC
    # CELERY_TIMEZONE='UTC'                             
    
    CELERY_IMPORTS = (                                  # 指定导入的任务模块
        'celery_app.task1',
        'celery_app.task2'
    )

**task1.py 代码如下：**

import time
    from celery_app import app
    
    @app.task
    def add(x, y):
        time.sleep(2)
        return x + y

**task2.py 代码如下：**

import time
    from celery_app import app
    
    @app.task
    def multiply(x, y):
        time.sleep(2)
        return x * y

**client.py 代码如下：**

# -*- coding: utf-8 -*-
    
    from celery_app import task1
    from celery_app import task2
    
    task1.add.apply_async(args=[2, 8])        # 也可用 task1.add.delay(2, 8)
    task2.multiply.apply_async(args=[3, 7])   # 也可用 task2.multiply.delay(3, 7)
    
    print 'hello world'

**现在，进入celery\_app 所在的目中，启动 Celery Worker 进程，在项目的根目录下执行下面命令：**

celery -A celery_app worker --loglevel=info

**若在后台运行，命令如下：**

nohup celery -A celery_app worker --loglevel=info &

**接着，运行 $ python client.py，它会发送两个异步任务到 Broker，在 Worker 的窗口我们可以看到如下输出：**

[2016-12-10 13:51:58,939: INFO/MainProcess] Received task: celery_app.task1.add[9ccffad0-aca4-4875-84ce-0ccfce5a83aa]
    [2016-12-10 13:51:58,941: INFO/MainProcess] Received task: celery_app.task2.multiply[64b1f889-c892-4333-bd1d-ac667e677a8a]
    [2016-12-10 13:52:00,948: INFO/PoolWorker-3] Task celery_app.task1.add[9ccffad0-aca4-4875-84ce-0ccfce5a83aa] succeeded in 2.00600231002s: 10
    [2016-12-10 13:52:00,949: INFO/PoolWorker-4] Task celery_app.task2.multiply[64b1f889-c892-4333-bd1d-ac667e677a8a] succeeded in 2.00601326401s: 21

--------------------

## 定时任务 ##

Celery 除了可以执行异步任务，也支持执行周期性任务（Periodic Tasks），或者说定时任务。**Celery Beat 进程**通过读取配置文件的内容，周期性地将定时任务发往任务队列。

**建立的项目结构如下：**

celery_demo                    # 项目根目录
        ├── celery_app             # 存放 celery 相关文件
            ├── __init__.py
            ├── celeryconfig.py    # 配置文件
            ├── task1.py           # 任务文件
            └── task2.py           # 任务文件

***\_init\_*.py 代码如下：**

# -*- coding: utf-8 -*-
    
    from celery import Celery
    
    app = Celery('demo')
    app.config_from_object('celery_app.celeryconfig')

**celeryconfig.py 代码如下：**

# -*- coding: utf-8 -*-
    
    from datetime import timedelta
    from celery.schedules import crontab
    
    # Broker and Backend
    BROKER_URL = 'redis://xxx.xxx.123.168:6379'         # 指定 Broker 即服务器或者本地的地址
    CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/0'  # 指定 Backend
    
    # Timezone
    CELERY_TIMEZONE='Asia/Shanghai'    # 指定时区，不指定默认为 'UTC'
    # CELERY_TIMEZONE='UTC'
    
    # import
    CELERY_IMPORTS = (
        'celery_app.task1',
        'celery_app.task2'
    )
    
    # schedules
    CELERYBEAT_SCHEDULE = {
        'add-every-30-seconds': {
             'task': 'celery_app.task1.add',
             'schedule': timedelta(seconds=30),       # 每 30 秒执行一次
             'args': (5, 8)                           # 任务函数参数
        },
        'multiply-at-some-time': {
            'task': 'celery_app.task2.multiply',
            'schedule': crontab(hour=9, minute=50),   # 每天早上 9 点 50 分执行一次
            'args': (3, 7)                            # 任务函数参数
        }
    }

**task1.py 代码如下：**

import time
    from celery_app import app
    
    @app.task
    def add(x, y):
        time.sleep(2)
        return x + y

**task2.py 代码如下：**

import time
    from celery_app import app
    
    @app.task
    def multiply(x, y):
        time.sleep(2)
        return x * y

**现在，进入celery\_app所在的目录中，启动 Celery Worker 进程，在项目的根目录下执行下面命令：**

celery -A celery_app worker --loglevel=info

**接着，启动 Celery Beat 进程，定时将任务发送到 Broker，在项目根目录下执行下面命令：**

celery beat -A celery_app

**显示结果如下：**

celery beat v4.0.1 (latentcall) is starting.
    __    -    ... __   -        _
    LocalTime -> 2016-12-11 09:48:16
    Configuration ->
        . broker -> redis://xxx.xxx.123.168:6379//
        . loader -> celery.loaders.app.AppLoader
        . scheduler -> celery.beat.PersistentScheduler
        . db -> celerybeat-schedule
        . logfile -> [stderr]@%WARNING
        . maxinterval -> 5.00 minutes (300s)

**之后，在 Worker 窗口我们可以看到，任务 task1 每 30 秒执行一次，而 task2 每天早上 9 点 50 分执行一次。**

**在上面，我们用两个命令启动了 Worker 进程和 Beat 进程，我们也可以将它们放在一个命令中：**

celery -B -A celery_app worker --loglevel=info

**若在后台运行，命令如下：**

nohup celery -B -A celery_app worker --loglevel=info &

Celery 周期性任务也有多个配置项，可参考[官方文档][Link 2]。

--------------------

# Celery Flower 监控工具 #

## flower 安装： ##

pip3 install flower

flower 是一个 celery 的监控工具，它提供了一个图形用户界面，可以极大的方便我们监控任务的执行过程， 执行细节及历史记录，还提供了统计功能。

## flower 启动： ##

用 ssh 登录阿里云端，进入 celery\_app 所在的目录中，运行如下的命令：

celery flower -A celery_app --broker=redis://xxx.xxx.123.168:6379//

## flower 访问： ##

http://xxx.xxx.123.168:5555/

**所有节点的状况**  
![这里写图片描述][SouthEast 2]

**所有task的状况**  
![这里写图片描述][SouthEast 3]

参考文档：[http://flower-docs-cn.readthedocs.io/zh/latest/index.html\#][http_flower-docs-cn.readthedocs.io_zh_latest_index.html]

[SouthEast]: /images/20210913/04581ee845eb4e35a103b80ba88ab81d.png
[SouthEast 1]: /images/20210913/6622ddd1db604b10b30e8e554d4c0279.png
[Link 1]: http://docs.celeryproject.org/en/latest/userguide/configuration.html
[Link 2]: http://docs.celeryproject.org/en/latest/userguide/periodic-tasks.html
[SouthEast 2]: /images/20210913/e95e85dd8a214b9a873e9430cc85be98.png
[SouthEast 3]: /images/20210913/76521183e7d44c94a3a396188b04aaad.png
[http_flower-docs-cn.readthedocs.io_zh_latest_index.html]: http://flower-docs-cn.readthedocs.io/zh/latest/index.html#