[ celery使用 ]

发表评论
94 views

Celery异步

官网 https://docs.celeryq.dev/en/stable/django/first-steps-with-django.html

-------------- celery@DESKTOP-6RC5C0D v5.2.3 (dawn-chorus)
--- ***** -----
-- ******* ---- Windows-10-10.0.19044-SP0 2022-04-01 15:34:29

- *** --- * ---

- ** ---------- [config]

- ** ---------- .> app:         meiduo:0x22dbfe08e50                    # 创建的对象名字，地址，

- ** ---------- .> transport:   redis://:**@192.168.10.128:6379/8       # 中间人，容器，地址

- ** ---------- .> results:     disabled://                           # 结果
  *** --- * --- .> concurrency: 4 prefork                               # 进程数量  
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)      
 -------------- [queues]                                                # 队列列表
 [tasks]
      . send_sms_code  任务队列

基本使用

安装

celery 安装
pip install —U Celery

windows 安装 
pip install eventlet

基本配置

结构目录

celery文件夹 与 manage.py 同级

image.png

main.py主文件

1. 在项目文件上一级创建celert_tasks文件
2. 在目录里面创建main.py文件夹
mian.py
from celery import Celery # 类里面
# 创建实例
celery_app = Celery('meiduo') # 标识 生产者

# 加载配置 容器在什么地方 config
celery_app.config_from_object('celery_tasks.config')

# 注册任务
celery_app.autodiscover_tasks(['celery_tasks.sms'])  # 注意：：：：是 列表


# 配置定时任务

celery_app.conf.timezone = 'Asia/Shanghai'   # 时区

celery_app.conf.enable_utc = False          # 是否使用UTC

celery_app.conf.beat_schedule = {           # 配置beat
    # 名字命名
    'add-every-10-seconds':{
        # 执行task下的 test_email函数
        'task':'test_email',
        # 每6秒执行一次
        # 'schedult': 6,

        # 每隔1分钟执行一次
        # schedule: crontab(minute="*/1")

        # 每6秒一次(时差=6s) 更丰富跟灵活
        'schedule':timedelta(seconds=6),
        # 传参
        'args':('张三',)
    }
}

config.py 配置文件

# celery 的配置文件

# 指定中间人、 消息队列、 任务队列、容器 、redis

# redis
result_backend = 'redis://:Ww550076@42.193.152.252:6379/4'  # (存储结果)

# redis
broker_url='redis://:123456@192.168.10.128:6379/8'  # 数据库消息中间件

# rbmq
broker_url = 'amqp://%s:%s@42.193.152.252:5672/blog' % ('admin', 'chenhaifeng1124')  # 林的服务器 (储存任务)

创建工作目录文件夹

tasks文件夹

  • 任务.py
  • init.py
from celery_tasks.sms.ronglian_sms_sdk.SendMessage import CCP
from celery_tasks.main import celery_app
# 使用装饰器异步任务，保证celery识别任务  celery 提供的装饰器
@celery_app.task(name='send_sms_code')  # 起别名
def send_sms_code(mobile,sms_code):
    '''发送短信验证码的异步任务'''
    '''
    mobile:手机号
    sms_code:短信验证码
    :return: 成功:0 失败:-1
    '''
    send_ret = CCP().send_template_sms('1', mobile, (sms_code, '5'))
    return send_ret

widows运行命令服务器celery

celery -A celery_tasks.main worker -l info -P eventlet 启动服务器
celery -A celery_tasks.main beat # 读取amin里的beat配置(定时任务)

python调用使用

执行异步任务

# 执行异步任务
from celery_tasks.email.tasks import send_verify_email # 导入celery里自定义的函数
import datetime
from celery_tasks.test.tasks import test_sms, test_email
#from celery_tasks.text_main import cel
#from celery.result import AsyncResult


# 执行异步任务  (使用函数.dealy("参数")格式调用)
# result = send_email.delay('陈海峰')
# print(result.id)  # 返回的id
#
# result = send_sms.delay('欧子君')
# print(result)

执行定时任务

方式1

在消费者中apply_async传参eta时间

# 执行定时任务
from celery_tasks.email.tasks import send_verify_email # 导入celery里自定义的函数


# 转换当前时间为utc标准
ctime = datetime.datetime.now()

# 转换成国标日期对象
utc_ctime = datetime.datetime.utcfromtimestamp(ctime.timestamp())

# 设置10秒
time_delay = timedelta(seconds=10)

# 添加当前时间+10秒
task_time = utc_ctime + time_delay

# 执行apply_async方法（与delay基本一直，就是能接收更多参数）
result = test_email.apply_async(['911',],eta=task_time)

print(result.id)

django视图中使用案例

# 转换当前时间为utc标准
def get(self, request):
    ctime = datetime.datetime.now()

    # 转换成国标日期对象
    utc_ctime = datetime.datetime.utcfromtimestamp(ctime.timestamp())

    # 设置10秒
    time_delay = datetime.timedelta(seconds=10)

    # 添加当前时间+10秒
    task_time = utc_ctime + time_delay
    print(task_time)
    # 定时任务celery
    result = test_celery.apply_async(['chenhaifeng',], eta=task_time)
    print(result.id)

    time.sleep(1)
    asy_res = celery_result(res_id=result.id, res_app=test_celery)
    print(asy_res)
    return JsonResponse(asy_res)

方式2

生产者配置的beat(定时任务)

celery -A celery_tasks.main worker -l info -P eventlet 启动服务器

# 并发量
-c 10

# 执行读取celery 配置文件 baet(配置的定时任务)
PS D:\Django_mark\New_blogger\new_blog> celery -A celery_task.main beat

# 终端打印值
celery beat v5.2.0 (dawn-chorus) is starting.
__    -    ... __   -        _
LocalTime -> 2022-11-15 12:25:19
Configuration ->
    . broker -> amqp://admin:**@42.193.152.252:5672/blog
    . loader -> celery.loaders.app.AppLoader
    . scheduler -> celery.beat.PersistentScheduler
    . db -> celerybeat-schedule
    . logfile -> [stderr]@%WARNING
    . maxinterval -> 5.00 minutes (300s)

获取异步结果

from celery.result import AsyncResult
from celery_task.main import celery_app


# 获取异步结果
async_result = AsyncResult(id='b065a154-b0e6-4b5b-848e-4de728295b91',app=cel)

def celery_result(res_id, res_app):
    # res_id为结果id, res_app为执行的任务函数
    async_result = AsyncResult(id=res_id, app=res_app)
    res = {
        'code': 400,
        'res': '执行失败',
        'status': '执行失败',
    }
    if async_result.successful():
        # 获取结果
        result = async_result.get()
        print(result)
        res['code'] = 0
        res['res'] = result
        res['status'] = '执行成功'
        # res,forget() # 将结果删除,执行完成,结果不会删除
        # async.revoke(terminate=true) # 无论显示是什么时候都要终止
        # async,revoke(terminate=false) # 如果任务还没开始执行, 那么就可以终止

    elif async_result.failed():
        print('执行失败')

    elif async_result.status == 'PENDING':
        res['code'] = 0
        res['res'] = async_result.status
        res['status'] = '任务等待'
        print('任务等待')

    elif async_result.status == 'RETRY':
        res['code'] = 0
        res['res'] = async_result.status
        res['status'] = '任务异常,正在重试'
        print('任务异常,正在重试')

    elif async_result.status == 'STARTED':
        res['code'] = 0
        res['res'] = async_result.status
        res['status'] = '任务已经开始被执行'
        print('任务已经开始被执行')
    return res

image.png

Django完整案例

结构

  • (项目根目录创建文件夹celery_task)
D:celery_task
│  config.py
│  main.py
│  result_celery.py
│
├─email
│  │  tasks.py
│  │  __init__.py
│  │
│  └─__pycache__
│          tasks.cpython-311.pyc
│          __init__.cpython-311.pyc
│
└─__pycache__
        config.cpython-311.pyc
        main.cpython-311.pyc
        result_celery.cpython-311.pyc

main.py

from celery import Celery
from datetime import timedelta
import os

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "o_project.settings")

celery_app = Celery('diary')

celery_app.config_from_object('celery_task.config')

celery_app.autodiscover_tasks(['celery_task.email'])

celery_app.conf.beat_schedule = {           # 配置beat
    'add-every-10-seconds':{
        # 执行task下的 test_email函数
        'task':'send_sms_code',
        # 每6秒执行一次
        # 'schedult': 6,

        # 每隔1分钟执行一次
        # schedule: crontab(minute="*/1")

        # 每6秒一次(时差=6s) 更丰富跟灵活
        'schedule':timedelta(seconds=6),
        # 传参
        'args':(17666292938,1)
    }
}

config.py

  • amqp任务队列
  • mongodb存储结果
# result_backend = "" # 储存结果
broker_url = "amqp://%s:%s@42.193.152.252:5672/blog" % ('admin', 'chenhaifeng1124') # 储存任务
result_backend = 'mongodb://%s:%s@192.168.50.225:27017/' % ('test', '123456')
mongodb_backend_settings = {
    'database': 'diary',
    'taskmeta_collection': 'diary_celery_collection',
}

result_celery.py

  • 查询结果
from celery.result import AsyncResult

# 获取异步结果
# async_result = AsyncResult(id='b065a154-b0e6-4b5b-848e-4de728295b91',app=cel)

def celery_result(res_id, res_app):
    async_result = AsyncResult(id=res_id, app=res_app)
    res = {
        'code': 400,
        'res': '执行失败',
        'status': '执行失败',
    }
    if async_result.successful():
        # 获取结果
        result = async_result.get()
        print(result)
        res['code'] = 0
        res['res'] = result
        res['status'] = '执行成功'
        # res,forget() # 将结果删除,执行完成,结果不会删除
        # async.revoke(terminate=true) # 无论显示是什么时候都要终止
        # async,revoke(terminate=false) # 如果任务还没开始执行, 那么就可以终止
        print('执行成功')

    elif async_result.failed():
        print('执行失败')

    elif async_result.status == 'PENDING':
        res['code'] = 0
        res['res'] = async_result.status
        res['status'] = '任务等待'
        print('任务等待')

    elif async_result.status == 'RETRY':
        res['code'] = 0
        res['res'] = async_result.status
        res['status'] = '任务异常,正在重试'
        print('任务异常,正在重试')

    elif async_result.status == 'STARTED':
        res['code'] = 0
        res['res'] = async_result.status
        res['status'] = '任务已经开始被执行'
        print('任务已经开始被执行')
    return res

创建任务

  • celery_task/email文件夹
  • 文件夹下创建tasks.py文件
  • 文件夹下创建__init__.py文件
from celery_task.main import celery_app

# 使用装饰器异步任务，保证celery识别任务  celery 提供的装饰器
@celery_app.task(name='send_sms_code')  # 起别名
def send_sms_code(mobile,sms_code):
    '''发送短信验证码的异步任务'''
    '''
    mobile:手机号
    sms_code:短信验证码
    :return: 成功:0 失败:-1
    '''

    print(f'执行异步任务send_sms_code{mobile}, {sms_code}')
    return mobile, sms_code

终端中测试调用任务

  • 开启celery
celery -A celery_task.main worker -l info -P eventlet
  • 测试任务delay
python manage.py shell
>>> from celery_task.email.tasks import send_sms_code
>>> from celery_task.result_celery import celery_result
>>> send_sms_code.delay(13113003318,1)  # 运行
>>> res = celery_result('30054694-b865-47dd-bd83-7a1973f29541', send_sms_code) # 获取储存结果
>>> res
{'code': 0, 'res': [13113003318, 1], 'status': '执行成功'}