准备工作
(1) 环境
@J1 server, CentOS7.9$ which gcc
/home/wangjl/soft/gcc-12.3.0/bin/gcc$ which python3
/home/wangjl/soft/python3/python-3.10.14/bin/python3$ gcc --version
gcc (GCC) 12.3.0
Copyright (C) 2022 Free Software Foundation, Inc.$ python3 -V
Python 3.10.14
(2) 准备
$ pip3 install celery
$ pip3 install redis
$ pip3 list | grep celery
celery 5.4.0
$ pip3 list | grep redis
redis 5.0.5
启动redis服务:
$ redis-server --version
Redis server v=6.2.6 sha=00000000:0 malloc=jemalloc-5.1.0 bits=64 build=3ee1339f93e3f95a
$ redis-server
…
76382:M 12 Jun 2024 16:54:15.813 * Ready to accept connections
启动服务后才能连接：
$ redis-cli -h 127.0.0.1 -p 6379
127.0.0.1:6379> ping
PONG
127.0.0.1:6379> quit
$
1. 定义任务
$ cat task02.py
from celery import Celery
import subprocess#app = Celery('tasks', broker='amqp://guest@localhost//')
app = Celery('tasks_02', broker='redis://localhost:6379/0', backend='redis://localhost:6379/1')@app.task
def run_command(cmd):return subprocess.check_output(cmd, shell=True)
解释：run_command是一个Celery任务，它接受一个命令字符串cmd并使用subprocess.check_output来执行它。
2. 启动任务
$ celery -A task02 worker --loglevel=info
注意，task02是py文件的文件名，不含py后缀名。
3. 在代码中异步执行Linux命令
$ python3
from task02 import run_command# 异步执行命令
result = run_command.apply_async(args=['ls -la'])# 如果需要，你可以获取结果
print(result.get())
4. 查看任务状态
>>> result.id
'588b3799-c424-464f-a2ef-125c87751643'
>>> result.status
'SUCCESS'也可以记录id，根据id查询任务执行状态：
from task02 import app
from celery.result import AsyncResult # 导入AsyncResult类
async_result = AsyncResult(id=result.id, app=app)>>> async_result
<AsyncResult: 588b3799-c424-464f-a2ef-125c87751643>
>>> async_result.status
'SUCCESS'
5. 异步执行耗时任务时，记录id到数据库，根据id查看状态，根据状态显示结果
# 异步执行耗时命令
result2 = run_command.apply_async(args=['sleep 10 && pwd && ls -la'])
task_id=result2.id #记录id到数据库
>>> task_id
'6fba0cfa-fa9a-46dd-b661-1708c55376b8'# 根据id获取对象
async_result = AsyncResult(id=task_id, app=app)
# 根据状态获取结果
if "SUCCESS" == async_result.status:print(async_result.get())
)
