文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

Python中celery的使用

2024-04-02 19:55

关注

 Celery简介

  Celery是一个简单、灵活且可靠的,处理大量消息的分布式系统,专注于实时处理的异步任务队列,同时也支持任务调度。

  Celery的架构由三部分组成,消息中间件(message broker),任务执行单元(worker)和任务执行结果存储(task result store)组成。

  消息中间件:Celery本身不提供消息服务,但是可以方便的和第三方提供的消息中间件集成。包括,RabbitMQ, Redis等等。

  任务执行单元:Worker是Celery提供的任务执行的单元,worker并发的运行在分布式的系统节点中。

  任务结果存储:Task result store用来存储Worker执行的任务的结果,Celery支持以不同方式存储任务的结果,包括AMQP, redis等。

  版本支持情况:


Celery version 4.0 runs on
        Python ❨2.7, 3.4, 3.5❩
        PyPy ❨5.4, 5.5❩
    This is the last version to support Python 2.7, and from the next version (Celery 5.x) Python 3.5 or newer is required.

    If you're running an older version of Python, you need to be running an older version of Celery:

        Python 2.6: Celery series 3.1 or earlier.
        Python 2.5: Celery series 3.0 or earlier.
        Python 2.4 was Celery series 2.2 or earlier.

    Celery is a project with minimal funding, so we don't support Microsoft Windows. Please don't open any issues related to that platform.

  Celery多用来执行异步任务,将耗时的操作交由Celery去异步执行,比如发送邮件、短信、消息推送、音视频处理等。还可以执行定时任务,定时执行某件事情,比如Redis中的数据每天凌晨两点保存至mysql数据库,实现Redis的持久化。

celery的异步任务

celery的使用

1.安装celery

$ pip install -U celery

1)安装相关依赖

$ pip install "celery[redis,auth,msgpack]"

序列化程序

celery[auth]

用于使用auth安全序列化程序。

celery[msgpack]

用于使用 msgpack 序列化程序。

celery[redis]

使用 Redis 作为消息传输或结果后端。

2.安装redis

这里我们使用redis作为celery的broker,作为任务队列的存储和结果的存储。

对于 Redis 支持,您必须安装其他依赖项。您可以使用celery[redis] bundle一次性安装 Celery 和这些依赖项:

$ pip install -U "celery[redis]"

1)配置

配置很简单,只需配置你的 Redis 数据库的位置:

app.conf.broker_url = 'redis://localhost:6379/0'

其中 URL 的格式为:

redis://:password@hostname:port/db_number

方案后面的所有字段都是可选的,并且将默认为localhost 端口 6379,使用数据库 0。

3.使用ceelry

1)首先我们可以创建一个celery的文件夹,然后创建一个tasks.py文件

celery/tasks.py


from celery import Celery

# 第一个参数就是当前脚本的名称,backend 任务执行结果的存储地址broker 任务队列的存储地址
app = Celery('tasks', backend='redis://127.0.0.1', broker='redis://127.0.0.1')

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

celery/run_tasks.py


from tasks import add

result = add.delay(1, 2)
print('Is task ready: %s' % result.ready())  # False说明任务还没有执行完
run_result = result.get(timeout=1)
print('task result: %s' % run_result)

print('Is task ready: %s' % result.ready())

4.启动celery


$ cd celry
$ celery -A tasks worker --loglevel=info

使用flower监控celery任务的执行情况

pip install flower

启动flower,指定我们的应用,确保你的celery是启动的。


cd celery
celery -A tasks flower --broker=redis://@localhost:6379/0

运行结果:

celery [celery args] flower [flower args].
[I 210825 10:54:00 command:152] Visit me at http://localhost:5555
[I 210825 10:54:00 command:159] Broker: redis://127.0.0.1:6379//
[I 210825 10:54:00 command:160] Registered tasks:

我们就可以通过5555端口看到celery异步任务的运行情况了

![image-20210825113106220](/Users/gelong/Library/Application Support/typora-user-images/image-20210825113106220.png)

Django中使用celery

官方地址:https://docs.celeryproject.org/en/latest/django/first-steps-with-django.html

1.创建celery文件

根据官方文档的说明,我们可以直接在Django项目同名的应用下创建celery.py文件

recruitment/recruitment/celery.py


import os

from celery import Celery

# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SEttINGS_MODULE', 'recruitment.base')  # 这里我把配置文件放到了根目录下的settings/base.py 中

app = Celery('recruitment')

# Using a string here means the worker doesn't have to serialize
# the configuration object to child processes.
# - namespace='CELERY' means all celery-related configuration keys
#   should have a `CELERY_` prefix.
app.config_from_object('django.conf:settings', namespace='CELERY')

# Load task modules from all registered Django apps.
app.autodiscover_tasks()

def debug_task(self):
  print(f'Request: {self.request!r}')

然后我们需要在这个celery.py文件所在的目录的__init__文件中添加:


from __future__ import absolute_import, unicode_literals


# This will make sure the app is always imported when/保证所有app下的任务都能导入进来
# Django starts so that shared_task will use this app.
from .celery import app as celery_app

__all__ = ('celery_app',)

2.添加celery配置

settings/base.py


CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/1'
CELERY_ACCEPT_CONTENT = ['application/json']
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TASK_SERIALIZER = 'json'
CELERY_TIMEZONE = 'Asia/Shanghai'
CELERYD_MAX_TASKS_PER_CHILD = 10
CELERYD_LOG_FILE = os.path.join(BASE_DIR, "logs", "celery_work.log")
CELERYBEAT_LOG_FILE = os.path.join(BASE_DIR, "logs", "celery_beat.log")

3.在别的应用下使用celery执行异步任务 [使用celery异步发送钉钉群消息通知]

1.首先我们需要在应用下创建一个tasks.py文件interview/tasks.py


from __future__ import absolute_import, unicode_literals

from celery import shared_task
from .dingtalk import send

@shared_task
def send_dingtalk_message(message):
    send(message)

interview/dingtalk.py


from dingtalkchatbot.chatbot import DingtalkChatbot

from django.conf import settings


def send(message, at_mobiles=[]):
    # 引用 settings里面配置的钉钉群消息通知的WebHook地址:
    webhook = settings.DINGTALK_WEB_HOOK

    # 初始化机器人小Y,
    xiaoY = DingtalkChatbot(webhook)

    # 方式二:勾选“加签”选项时使用(v1.5以上新功能)
    # xiaoY = DingtalkChatbot(webhook, secret=secret)

    # Text消息@所有人
    xiaoY.send_text(msg=('消息通知: %s' % message), at_mobiles=at_mobiles)

interview.views.py


from interview.tasks import send_dingtalk_message

def notify_interview(modeladmin, request, queryset):
    candidates = ''
    interviewers = ''
    for obj in queryset:
        candidates = obj.userame + '' + candidates
        interviewers = obj.first_interviewer_user + '' + interviewers
    # 这里的消息发送到钉钉, 或者通过 Celery 异步发送到钉钉
    send_dingtalk_message.delay('候选人 %s 进入面试环节, 亲爱的面试官请做好面试准备:%s。' % (candidates, interviewers))

4.启动celery服务

启动celery服务,到我们的项目根目录启动,然后执行


$ celery -A recruitment worker -l info

如果需要制定配置文件,如果在mac下可以执行:


$ DJANGO_SEttINGS_MODULE=settings.base celery --app=recruitment worker --loglevel=info

启动flower监控异步任务


$ celery -A recruitment flower --broker=redis://localhost:6379/0

celery定时任务

到此这篇关于Python中celery的使用的文章就介绍到这了,更多相关celery的使用内容请搜索编程网以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程网!

阅读原文内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     807人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     351人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     314人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     433人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     221人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-后端开发
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯