【Celery】Celery的简易部署和应用

Celery的简易应用

---

文章目录

• Celery的简易应用
• 前言
• 一、创建一个Celery App
• 二、部署Celery
• • 1. 创建所需的用户和文件夹
  • 2. 创建 celeryd.config 配置文件
  • 3. 创建 .service 服务脚本
  • • (1) 创建路由任务 celeryd.service
    • (2) 创建周期任务 celery_beat.service
    • (3) 启动
• 三. 任务
• • • (1) 路由任务(routing task)
    • (2) 周期任务(periodic task)

---

前言

   Celery 是一个简单、灵活、可靠的分布式系统，可以处理大量的消息，同时提供维护系统所需的工具。它是一个专注于处理 实时任务 和 定时任务 的的任务队列。
   本篇文章使用 Ubuntu20.04 系统 ，Python3.8 版本。使用的是 Celery4.4.7 版本，用 Redis 作为消息队列存储任务（可以用更好的RabbitMQ）。
   主要目的是快速部署和简易应用。更多 *** 作：传送门 - Celery官方文档

---

一、创建一个Celery App

假设在下面目录里创建一个 Celery App，目录结构如下

目录 /home/ubuntu/app/Celery

# app.py
from celery import Celery

# include代表任务文件
app = Celery(broker='redis://localhost:6379/0',
             backend='redis://localhost:6379/1',
             # 导入所执行的任务文件
             include=['routing_task.tasks', 'periodic_task.tasks'])
             
# 配置时区，定时任务要用
app.conf.update(
    result_expires=3600,
    timezone='Asia/Shanghai'
)

# 定时任务调度
app.conf.beat_schedule = {
	# 任务名
    'this is periodic_tasks': {
    	# 任务调度位置
        'task': 'periodic_tasks.tasks.abbc',
        # 调度时间
        'schedule': timedelta(seconds=5),
        # 'schedule': crontab(minute=53, hour=14),
        # 任务参数
        'args': (16, 16),
    },
}

#　routing_task/tasks.py
from app import app


@app.task()
def add(x, y):
    return x + y


if __name__ == "__main__":
    print(add.delay(1, 3).result)

此时，我们可以在当前目录下，运行：

celery -A app worker --loglevel=INFO

就可以正常运行Celery了。
执行 python tasks.py，将add任务推入redis消息中间件后Celery执行它并返回处理结果。

---

二、部署Celery

使用 .service 守护进程来运行 Celery

1. 创建所需的用户和文件夹

# 创建 用户、组
sudo useradd celery
sudo groupadd celery
sudo gpasswd -a celery celery

# 创建日志目录
sudo mkdir /var/log/celery
sudo chmod 0755 /var/log/celery/
sudo chown celery:celery /var/run/celery/

# 创建pid文件夹
sudo mkdir /run/celery 
sudo chmod 0755 /run/celery/

2. 创建 celeryd.config 配置文件

配置文件路径：/etc/celery/celeryd.conf，没有就创建一个

# 节点的启动名称
CELERYD_NODES="work"
# 或者你可以写三个:
#CELERYD_NODES="w1 w2 w3"

# celery的绝对路径，还是别用相对路径了，避免出错。
CELERY_BIN="/usr/local/bin/celery"
#CELERY_BIN="/virtualenvs/def/bin/celery"

# 这里写Celery的应用实例
CELERY_APP="app"
# 或者你想写一个完整版
#CELERY_APP="Celery.app:app"

# 启动多个
CELERYD_MULTI="multi"

# worker的参数 , 启动8个
CELERYD_OPTS="--time-limit=300 --concurrency=8"
# celerybeat调度器路径
CELERYBEAT_OPTS="--schedule=/var/run/celery/celerybeat-schedule"

# 下面就是celery的pid路径，log日志了。
# - %n will be replaced with the first part of the nodename.
# - %I will be replaced with the current child process index
#   and is important when using the prefork pool to avoid race conditions.
CELERYD_PID_FILE="/var/run/celery/%n.pid"
CELERYD_LOG_FILE="/var/log/celery/%n%I.log"
CELERYD_LOG_LEVEL="INFO"

# 设置celerybeat的pid和日志文件路径
CELERYBEAT_PID_FILE="/var/run/celery/beat.pid"
CELERYBEAT_LOG_FILE="/var/log/celery/beat.log"

3. 创建 .service 服务脚本 (1) 创建路由任务 celeryd.service

# 路径 vim /etc/systemd/system/celeryd.service 
[Unit]
Description=Celery Service
After=network.target

[Service]
Type=forking
User=celery
Group=celery
# 配置文件的路径
EnvironmentFile=/etc/celery/celeryd.config
# Celery的工作路径
WorkingDirectory=/home/ubuntu/app/Celery
# 下面这写会被config里的参数自动替换掉
ExecStart=/bin/sh -c '${CELERY_BIN} -A $CELERY_APP multi start $CELERYD_NODES \
    --pidfile=${CELERYD_PID_FILE} --logfile=${CELERYD_LOG_FILE} \
    --loglevel="${CELERYD_LOG_LEVEL}" $CELERYD_OPTS'
ExecStop=/bin/sh -c '${CELERY_BIN} multi stopwait $CELERYD_NODES \
    --pidfile=${CELERYD_PID_FILE} --logfile=${CELERYD_LOG_FILE} \
    --loglevel="${CELERYD_LOG_LEVEL}"'
ExecReload=/bin/sh -c '${CELERY_BIN} -A $CELERY_APP multi restart $CELERYD_NODES \
    --pidfile=${CELERYD_PID_FILE} --logfile=${CELERYD_LOG_FILE} \
    --loglevel="${CELERYD_LOG_LEVEL}" $CELERYD_OPTS'
Restart=always

[Install]
WantedBy=multi-user.target

(2) 创建周期任务 celery_beat.service

[Unit]
Description=Celery Beat Service
After=network.target

[Service]
Type=simple
User=celery
Group=celery
EnvironmentFile=/etc/celery/celeryd.config
WorkingDirectory=/home/ubuntu/app/Celery
ExecStart=/bin/sh -c '${CELERY_BIN} -A ${CELERY_APP} beat  \
    --pidfile=${CELERYBEAT_PID_FILE} \
    --logfile=${CELERYBEAT_LOG_FILE} --loglevel=${CELERYD_LOG_LEVEL} $CELERYBEAT_OPTS'
Restart=always

[Install]
WantedBy=multi-user.target

(3) 启动

# 重载所有服务文件
sudo systemctl daemon-reload

# 启动 celeryd.service
sudo systemctl start celeryd.service
# 启动 celery_beat.service
sudo systemctl start celery_beat.service

三. 任务 (1) 路由任务(routing task)

#　routing_task/tasks.py
from app import app


@app.task()
def add(x, y):
    return f'this is routing task: {x - y}'

在所需要的场景下，通过

res = add.delay(1, 3)

来应用。

(2) 周期任务(periodic task)

在 app.conf.beat_schedule 中注册后， 重启 celery_beat 和 celeryd 两个服务后可以实现周期任务的添加应用。
注意：celery_beat 只当作任务调度，不执行任务。具体执行需要 celeryd 来完成，所以添加/修改任务后需要将两个 service 都重启才可以。
注册

# app.py
app.conf.beat_schedule = {
	# 任务名
    'this is periodic_tasks': {
    	# 任务调度位置
        'task': 'periodic_tasks.tasks.abbc',
        # 调度时间
        'schedule': timedelta(seconds=5),
        # 'schedule': crontab(minute=53, hour=14),
        # 任务参数
        'args': (16, 16),
    },
    ....
}

# periodic_task/tasks.py

from app import app

@app.task()
def abc(x, y):
    return f'this is periodic task: {x - y}'