celery-定时更新缓存数据

时间  2019-11-13
标签 celery 定时 更新 缓存 数据 繁體版
原文   原文链接

目录html

Celery

官方

Celery 官网python

Celery 官方文档英文版redis

Celery 官方文档中文版django

Celery架构

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

消息中间件

Celery自己不提供消息服务，可是能够方便的和第三方提供的消息中间件集成。包括，RabbitMQ, Redis等等缓存

任务执行单元

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

任务结果存储

Task result store用来存储Worker执行的任务的结果，Celery支持以不一样方式存储任务的结果，包括AMQP, redis等并发

使用场景

异步任务：将耗时操做任务提交给Celery去异步执行，好比发送短信/邮件、消息推送、音视频处理等等app

定时任务：定时执行某件事情，好比天天数据统计异步

Celery的安装配置

pip install celery

消息中间件：RabbitMQ/Redis

app=Celery('任务名', broker='xxx', backend='xxx')

Celery执行异步任务

包架构封装

project
    ├── celery_task     # celery包
    │   ├── __init__.py # 包文件
    │   ├── celery.py   # celery链接和配置相关文件，且名字必须交celery.py
    │   └── tasks.py    # 全部任务函数
    ├── add_task.py     # 添加任务
    └── get_result.py   # 获取结果

基本使用

celery.py
# 1）建立app + 任务

# 2）启动celery(app)服务：
# 非windows
# 命令：celery worker -A celery_task -l info
# windows：
# pip3 install eventlet
# celery worker -A celery_task -l info -P eventlet

# 3）添加任务：手动添加，要自定义添加任务的脚本，右键执行脚本

# 4）获取结果：手动获取，要自定义获取任务的脚本，右键执行脚本


from celery import Celery

broker = 'redis://127.0.0.1:6379/1'
backend = 'redis://127.0.0.1:6379/2'
app = Celery(broker=broker, backend=backend, include=['celery_task.tasks'])
tasks.py
from .celery import app
import time
@app.task
def add(n, m):
    print(n)
    print(m)
    time.sleep(10)
    print('n+m的结果：%s' % (n + m))
    return n + m

@app.task
def low(n, m):
    print(n)
    print(m)
    print('n-m的结果：%s' % (n - m))
    return n - m
add_task.py
from celery_task import tasks

# 添加当即执行任务
t1 = tasks.add.delay(10, 20)
t2 = tasks.low.delay(100, 50)
print(t1.id)


# 添加延迟任务
from datetime import datetime, timedelta
def eta_second(second):
    ctime = datetime.now()
    utc_ctime = datetime.utcfromtimestamp(ctime.timestamp())
    time_delay = timedelta(seconds=second)
    return utc_ctime + time_delay

tasks.low.apply_async(args=(200, 50), eta=eta_second(10))
get_result.py
from celery_task.celery import app

from celery.result import AsyncResult

id = '21325a40-9d32-44b5-a701-9a31cc3c74b5'
if __name__ == '__main__':
    async = AsyncResult(id=id, app=app)
    if async.successful():
        result = async.get()
        print(result)
    elif async.failed():
        print('任务失败')
    elif async.status == 'PENDING':
        print('任务等待中被执行')
    elif async.status == 'RETRY':
        print('任务异常后正在重试')
    elif async.status == 'STARTED':
        print('任务已经开始被执行')

高级使用

celery.py
# 1）建立app + 任务

# 2）启动celery(app)服务：
# 非windows
# 命令：celery worker -A celery_task -l info
# windows：
# pip3 install eventlet
# celery worker -A celery_task -l info -P eventlet

# 3）添加任务：自动添加任务，因此要启动一个添加任务的服务
# 命令：celery beat -A celery_task -l info

# 4）获取结果


from celery import Celery

broker = 'redis://127.0.0.1:6379/1'
backend = 'redis://127.0.0.1:6379/2'
app = Celery(broker=broker, backend=backend, include=['celery_task.tasks'])


# 时区
app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
app.conf.enable_utc = False

# 任务的定时配置
from datetime import timedelta
from celery.schedules import crontab
app.conf.beat_schedule = {
    'low-task': {
        'task': 'celery_task.tasks.low',
        'schedule': timedelta(seconds=3),
        # 'schedule': crontab(hour=8, day_of_week=1),  # 每周一早八点
        'args': (300, 150),
    }
}
tasks.py
from .celery import app

import time
@app.task
def add(n, m):
    print(n)
    print(m)
    time.sleep(10)
    print('n+m的结果：%s' % (n + m))
    return n + m


@app.task
def low(n, m):
    print(n)
    print(m)
    print('n-m的结果：%s' % (n - m))
    return n - m
get_result.py
from celery_task.celery import app

from celery.result import AsyncResult

id = '21325a40-9d32-44b5-a701-9a31cc3c74b5'
if __name__ == '__main__':
    async = AsyncResult(id=id, app=app)
    if async.successful():
        result = async.get()
        print(result)
    elif async.failed():
        print('任务失败')
    elif async.status == 'PENDING':
        print('任务等待中被执行')
    elif async.status == 'RETRY':
        print('任务异常后正在重试')
    elif async.status == 'STARTED':
        print('任务已经开始被执行')

django中使用

celery.py
# 重点：要将 项目名.settings 所占的文件夹添加到环境变量
# import sys
# sys.path.append(r'项目绝对路径')

# 开启django支持
import os
os.environ.setdefault('DJANGO_SETTINGS_MODULE', '项目名.settings')
import django
django.setup()



# 1）建立app + 任务

# 2）启动celery(app)服务：
# 非windows
# 命令：celery worker -A celery_task -l info
# windows：
# pip3 install eventlet
# celery worker -A celery_task -l info -P eventlet

# 3）添加任务：自动添加任务，因此要启动一个添加任务的服务
# 命令：celery beat -A celery_task -l info

# 4）获取结果


from celery import Celery

broker = 'redis://127.0.0.1:6379/1'
backend = 'redis://127.0.0.1:6379/2'
app = Celery(broker=broker, backend=backend, include=['celery_task.tasks'])


# 时区
app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
app.conf.enable_utc = False

# 任务的定时配置
from datetime import timedelta
from celery.schedules import crontab
app.conf.beat_schedule = {
    'django-task': {
        'task': 'celery_task.tasks.test_django_celery',
        'schedule': timedelta(seconds=3),
        'args': (),
    }
}
tasks.py
from .celery import app

from home.models import Banner
from settings.const import BANNER_COUNT  # 轮播图最大显示条数
from home.serializers import BannerModelSerializer
from django.core.cache import cache
@app.task
def update_banner_list():
    # 获取最新内容
    banner_query = Banner.objects.filter(is_delete=False, is_show=True).order_by('-orders')[:BANNER_COUNT]
    # 序列化
    banner_data = BannerModelSerializer(banner_query, many=True).data
    for banner in banner_data:
        banner['image'] = 'http://127.0.0.1:8000' + banner['image']
    # 更新缓存
    cache.set('banner_list', banner_data)
    return True
相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程