码迷,mamicode.com
首页 > 编程语言 > 详细

python3+celery+redis实现异步任务

时间:2020-02-08 09:43:16      阅读:148      评论:0      收藏:0      [点我收藏+]

标签:efault   自动   返回   消息   end   return   flow   order   send   

一、原理

Celery是基于Python开发的一个分布式任务队列框架,支持使用任务队列的方式在分布的机器/进程/线程上执行任务调度。它是Python写的库,但是它实现的通讯协议也可以使用ruby,php,javascript等调用。异步任务除了消息队列的后台执行的方式,还是一种则是定时计划任务。

Celery 是一个强大的分布式任务队列,它可以让任务的执行完全脱离主程序,甚至可以被分配到其他主机上运行。我们通常使用它来实现异步任务(async task)和定时任务(crontab)。它的架构组成如下图 

技术图片

 

 

 

组件:

1、任务(tasks)--用户定义的函数,用于实现用户的功能,比如执行一个耗时很长的任务

2、中间介(Broker)--用于存放tasks的地方,但是这个中间介需要解决一个问题,就是可能需要存放非常非常多的tasks,而且要保证Worker能够从这里拿取

3、执行者(Worker)--用于执行tasks,也就是真正调用我们在tasks中定义的函数

4、存储(Backend)--把执行tasks返回的结果进行存储,以供用户查看或调用

二、实现过程

1.环境安装(RabbitMQ/Redis、Celery、django-celery、flower)

2.创建工程

技术图片

 

 

 红圈为本工程所需:

web_order下面需要修改的文件:celery.py、__init__.py、settings文件

web_test下面需要修改的文件:tasks.py文件、longTask.py文件

3.修改文件过程

1)修改settings.py。在settings的最后加上如下代码:

1 # CELERY SETTING
2 BROKER_URL = redis://localhost:6379    #指定消息中间件
3 CELERY_RESULT_BACKEND = redis://localhost:6379   ##指定结果存储位置为本地数据库
4 CELERY_ACCEPT_CONTENT = [application/json]  #
5 CELERY_TASK_SERIALIZER = json
6 CELERY_RESULT_SERIALIZER = json
7 CELERY_TIMEZONE = Asia/Shanghai
8 CELERY_IMPORTS = ("web_test.tasks")   #注册任务,

2)__init__ 文件。

#绝对导入,以免celery和标准库中的celery模块冲突
from __future__ import absolute_import

# This will make sure the app is always imported when
# Django starts so that shared_task will use this app.

#以下导入时为了确保在Django启动时加载app,shared_task在app中会使用到
from .celery import app as celery_app

__all__ = [celery_app]

3)celery文件

from __future__ import absolute_import,unicode_literals
import os
from celery import Celery
from django.conf import settings

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "web_order.settings")  # 设置celery可以在命令行中使用
app = Celery(web_order)  # 创建app实例
# app = Celery(‘tcelery‘, backend=‘redis://localhost:6379/0‘, broker=‘redis://localhost:6379/0‘)
app.conf.CELERY_IGNORE_RESULT = False  # 结果不忽略
# app.conf.CELERY_RESULT_BACKEND = ‘redis://localhost:6379/0‘ #结果保存在redis中

app.config_from_object(django.conf:settings)  # 从文件中加载实例
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)  # 自动加载tasks,注意:他会去app下面查找tasks.py文件,所以我们必须将task放在tasks.py文件中

4)longTasks.py

from django.http import JsonResponse
import json
from .tasks import test   #导入异步任务的方法


def sendlongtask(request):
    #由此处调用test方法,执行异步命令
    run_res = test(sh, userIp, username)
    return JsonResponse("执行成功", safe=False)

5)Tasks.py

from celery import shared_task

@shared_task
def test(x, y):
    return (x+y)

4.启动Django和celery

  在项目根目录执行:

  python manage.py runserver 0.0.0.0:8000

  python manage.py celery worker -c 4 --loglevel=info

 

5.另外,Celery提供了一个工具flower,将各个任务的执行情况、各个worker的健康状态进行监控并以可视化的方式展现,如下图所示:

技术图片

 

 

 

 

    Django下实现的方式如下: 

      1.) 安装flower:

    pip install flower

      2.) 启动flower(默认会启动一个webserver,端口为5555):

python manage.py celery flower

     3.) 进入http://localhost:5555即可查看。

6.python3踩过的坑:python3.7与celery不兼容

技术图片

 

 1.出现这个错误时,需要将报错文件中所有的async改为asynchronous或者其他变量名即可。

执行下面的脚本也可快速修改

TARGET=/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/celery/backends
cd $TARGET
if [ -e async.py ]
then
    mv async.py asynchronous.py
    sed -n s/async/asynchronous/g redis.py
    sed -n s/async/asynchronous/g rpc.py
fi

运行后,你会发现celery可以正常使用了

2.将python版本降到3.6及以下,celery也可正常使用。

 

python3+celery+redis实现异步任务

标签:efault   自动   返回   消息   end   return   flow   order   send   

原文地址:https://www.cnblogs.com/weisunblog/p/12275261.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!