pip3 install celery

# celery_task.py

from celery import Celery

"""
# 如果redis没有设置密码
broker = ''redis://127.0.0.1:6379/1''
backend = ''redis://127.0.0.1:6379/2''
"""
broker = ''redis://:12345@127.0.0.1:6379/1''
backend = ''redis://:12345@127.0.0.1:6379/2''
# c1是实例化产生的celery的名字，因为会存在多个celery
app = Celery(''c1'', broker=broker, backend=backend)

# 需要使用一个装饰器，来管理该任务(函数)
@app.task
def add(x, y):
    import time
    time.sleep(1)
    return x + y

# submit.task.py

# 用于提交任务
from celery_task import add
# 提交任务到消息队列中，这里只是将任务提交，并没有执行

res = add. delay(3, 8)
print(res)
# 结果是标识任务的字符串(id号)
# 7811a028-428c-4dd5-9135-788e26e694a7

linux: celery worker -A celery_task -l info   
windows下：celery worker -A celery_task -l info -P eventlet

from celery.result import AsyncResult
from celery_task import app

async = AsyncResult(id=''bd600820-9366-4220-a679-3e435ae91e71'', 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的使用
    -pip3 install celery
    -写一个py文件：celery_task
        -1 指定broker（消息中间件），指定backend（结果存储）
        -2 实例化产生一个Celery对象 app=Celery(''名字''，broker，backend)
        -3 加装饰器绑定任务，在函数（add）上加装饰器app.task
        -4 其他程序提交任务,先导入add，add.delay(参，参数)，会将该函数提交到消息中间件，但是并不会执行，有个返回值，直接print会打印出任务的id，以后用id去查询任务是否执行完成
        -5 启动worker去执行任务：
        linux: celery worker -A celery_task_s1 -l info   
        windows下：celery worker -A celery_task_s1 -l info -P eventlet
        -6 查看结果：根据id去查询
            async = AsyncResult(id="bd600820-9366-4220-a679-3e435ae91e71", app=app)
            if async.successful():
                #取出它return的值
                result = async.get()
                print(result)

# celery的多任务结构
    -项目结构：
        pro_cel
            ├── celery_task# celery相关文件夹
            │   ├── celery.py   # celery连接和配置相关文件,必须叫这个名字
            │   └── tasks1.py    #  所有任务函数
            │   └── tasks2.py    #  所有任务函数
            ├── check_result.py # 检查结果
            └── send_task.py    # 触发任务
    -启动worker，celery_task是包的名字
        celery worker -A celery_task -l info -P eventlet

# celery.py

from celery import Celery
broker = ''redis://:12345@127.0.0.1:6379/1''
backend = ''redis://:12345@127.0.0.1:6379/2''
# c1是实例化产生的celery的名字，因为会存在多个celery
app = Celery(''c1'', broker=broker, backend=backend,
             # 包含一些2个任务文件，去相应的py文件找任务，对多个任务进行分类
             include=[
                 ''celery_task.task_1'',
                 ''celery_task.task_2'',
             ])


# celery提供一些配置，具体可查看官方文档
# app.conf.timezone = ''Asia/Shanghai''

# task_1.py
from celery_task.celery import app

@app.task
def add1(x, y):
    import time
    time.sleep(0.5)
    return x + y

# task_2.py
from celery_task.celery import app

@app.task
def add2(x, y):
    import time
    time.sleep(1)
    return x * y

# send_task.py
from celery_task.task_1 import add1
from celery_task.task_2 import add2


res1 = add1.delay(3, 8)
print(res1)   # 16e847f3-fc14-4391-89e2-e2b3546872cf

res2 = add2.delay(4, 9)
print(res2)   # 858c0ae5-8516-4473-8be5-7501fb856ff4

# check_reslut.py
from celery.result import AsyncResult
from celery_task.celery import app

for i in [''16e847f3-fc14-4391-89e2-e2b3546872cf'', ''858c0ae5-8516-4473-8be5-7501fb856ff4'']:
    async = AsyncResult(id=i, 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(''任务正在执行'')

# send_task.py

from celery_task.task_1 import add1
from celery_task.task_2 import add2
# 执行定时任务，3s以后执行add1、add2任务
from datetime import datetime
# 设置任务执行时间2019年7月12日21点45分12秒
v1 = datetime(2019, 7, 12, 21, 48, 12)
print(v1)  # 2019-07-12 21:45:12
# 将v1时间转成utc时间
v2 = datetime.utcfromtimestamp(v1.timestamp())
print(v2)  # 2019-07-12 13:45:12
# 取出要执行任务的时间对象，调用apply_async方法，args是任务函数传的参数，eta是执行的时间
result1 = add1.apply_async(args=[3, 8], eta=v2)
result2 = add2.apply_async(args=[4, 9], eta=v2)
print(result1.id)
print(result2.id)

# send_task.py

# 方式二：实际上和方法一类似，多了一个延迟时间，也就是用现在时间和推迟执行的时间计算出任务执行的最终utc时间
# 然后也是调用apply_async方法。
from datetime import datetime
ctime = datetime.now()
# 默认使用utc时间
utc_ctime = datetime.utcfromtimestamp(ctime.timestamp())
from datetime import timedelta
# 使用timedelta模块，拿到10秒后的时间对象，这里参数可以传秒、毫秒、微秒、分、小时、周、天
time_delay = timedelta(seconds=10)
# 得到任务运行时间：
task_time = utc_ctime + time_delay
result1 = add1.apply_async(args=[3, 8], eta=task_time)
result2 = add2.apply_async(args=[4, 9], eta=task_time)
print(result1.id)
print(result2.id)

# celery.py中

from celery import Celery
broker = ''redis://:12345@127.0.0.1:6379/1''
backend = ''redis://:12345@127.0.0.1:6379/2''
# c1是实例化产生的celery的名字，因为会存在多个celery
app = Celery(''c1'', broker=broker, backend=backend,
             # 包含一些2个任务文件，去相应的py文件找任务，对多个任务进行分类
             include=[
                 ''celery_task.task_1'',
                 ''celery_task.task_2'',
                 ''celery_task.task_3'',
             ])


# celery提供一些配置，具体可查看官方文档
# app.conf.timezone = "Asia/Shanghai"
# app.conf.enable_utc = True


# 计划任务
from datetime import timedelta
from celery.schedules import crontab
app.conf.beat_schedule = {
    ''submit_every_2_seconds'': {
        # 计划的任务执行函数
        ''task'': ''celery_task.task_1.add1'',
        # 每个2秒执行一次
        ''schedule'': timedelta(seconds=2),
        # 传递的任务函数参数
        ''args'': (3, 9)
    },
    ''submit_every_3_seconds'': {
            # 计划的任务执行函数
            ''task'': ''celery_task.task_2.add2'',
            # 每个3秒执行一次
            ''schedule'': timedelta(seconds=3),
            # 传递的任务函数参数
            ''args'': (4, 7)
    },
    ''submit_in_fix_datetime'': {
        ''task'': ''celery_task.task_3.add3'',
        # 比如每年的7月13日10点53分执行
        # 注意：默认使用utc时间，当前的时间中的小时必须要-8个小时才会到点提交
        ''schedule'': crontab(minute=53, hour=2, day_of_month=13, month_of_year=7),
        
        ''''''
        # 如果不想-8，可以先设置时区，再按正常时间设置
        app.conf.timezone = "Asia/Shanghai"
        app.conf.enable_utc = True
        ''''''
        ''args'': (''Hello World'',)
    }

}

# 上面写完后，需要起一个进程，启动计划任务
# celery beat -A celery_task -l info

# 启动worker：
# celery worker -A celery_task -l info -P eventlet

import os
# 加载Django环境，bbs是所在的Django项目名称
    os.environ.setdefault("DJANGO_SETTINGS_MODULE", ''bbs.settings'')
    # 引入Django模块
    import django
    # 初始化Django环境
    django.setup()
    # 从app当中导入models
    from app01 import models
    # 调用操作，拿到数据库中的所有Book数据对象
    books = models.Books.objects.all()

from celery import Celery
import time

my_task=Celery("task",broker="redis://:123456@127.0.0.1:6379",backend="redis://:123456@127.0.0.1:6379")

@my_task.task
def my_func1():
    time.sleep(10)
    return "任务1"

@my_task.task
def my_func2():
    return "任务2"


@my_task.task
def my_func3():
    return "任务"

from s1 import my_func1

res=my_func1.delay()
print(res)

from celery.result import AsyncResult
from s1 import my_task

async_task=AsyncResult(id="48029f4f-769e-438b-ac97-e89cc0bb1157",app=my_task)
# result=async_task.get()

if async_task.successful():
    result=async_task.get()
    print(result+"OK!")
else:
    print("任务还未执行完成！")