第1步：安装celery与redis

Mac OS X 安装

$ brew install Redis    # Redis 安装，需提前安装 brew
$ pip install celery    # pip 安装celery库
$ pip install redis     # pip 安装 redis库

如果您没有安装 brew，建议您先安装 brew 参考 brew安装

Windows 安装
参考菜鸟教程-Redis 安装
下载地址：https://github.com/MSOpenTech/redis/releases。

Redis 支持 32 位和 64 位。这个需要根据你系统平台的实际情况选择，这里我们下载 Redis-x64-xxx.zip压缩包到 C 盘，解压后，将文件夹重新命名为 redis。

第2步：编写代码

1在settings.py中加入

CELERY_BROKER_URL = 'redis://localhost'

#: Only add pickle to this list if your broker is secured
#: from unwanted access (see userguide/security.html)
CELERY_ACCEPT_CONTENT = ['json']
CELERY_RESULT_BACKEND = 'redis://localhost'
CELERY_TASK_SERIALIZER = 'json'

2编写任务：

import time
from celery import Celery
app = Celery('tasks', broker='redis://localhost', backend='redis://localhost')

@app.task
def add(x, y):
    time.sleep(15)   #使用sleep表示耗时操作
    print(2)
    return x+y

3在视图中调用tasks

from django.http import HttpResponse
from .tasks import add, add2
def testCelery(request):
    print(add.delay(4, 4))
    print(add2.delay(5, 4))
    print('成功')
    return HttpResponse('成功')

4在url中注册接口

from django.contrib import admin
from django.urls import path
from testCelery.views import testCelery

urlpatterns = [
    path('admin/', admin.site.urls),
    path('testCelery/', testCelery)
]

第3步：启动redis、celery、django

1：启动redis
方法1：
切换到redis安装目录下执行：redis-server.exe redis.windows.conf
方法2：
输入redis-server &，再打开一个终端窗口，输入：redis-cli ，得到 127.0.0.1:6379>，这个时候就可以进行reids存储操作了。
输入 ping 得到 PONG，代表Redis服务器已经正常安装、连接成功，可以使用了。
使用：redis-cli shutdown就可以关闭Redis。
Redis 命令参考：
http://doc.redisfans.com

2：启动celery
执行celery -A project worker -l debug

3：启动django

第4步：调用接口

实现异步

第5步：实现flower 对celery进行监控

1: 安装

pip install flower

2:启动flower
假设在server2上启动flower，flower默认的端口是5555.

celery flower -A 项目名字 --address=服务器ip --port=5555

3:监控界面
在浏览器上输入 http://server2_ip:5555, 可以看到界面

多线程task

在celery中是可以执行多线程的，下面来演示一下执行的示例。

import threading
from time import sleep,ctime
 
def smoke():
    for i in range(3):
        print("smoke...%d"%i)
        sleep(10)
 
def drunk():
    for i in range(3):
        print("drink...%d"%i)
        sleep(10)
 
 
@celery_app.task
def test_use_thread():
    print('---开始---:%s' % ctime())
 
    t1 = threading.Thread(target=smoke)
    t2 = threading.Thread(target=drunk)
 
    t1.start()
    t2.start()
 
    while True:
        length = len(threading.enumerate())
        print("当前运行的线程数为:%d" % length)
        if length <= 1:
            break
 
        sleep(0.5)
 
    # sleep(5) #
    print('---结束---:%s' % ctime())

在django项目的交互模式下执行多线程的task任务

In [1]: from celery_tasks.tasks import test_use_thread
 
In [2]: test_use_thread.delay()
Out[2]: <AsyncResult: 69bba7b0-3808-47f1-ad36-2fea75930901>

可以看到任务顺利并发执行，也就是Celery虽然自身已经是异步任务，还是可以继续在task中使用多线程的。

转载
https://www.jianshu.com/p/61e573611a06?tdsourcetag=s_pcqq_aiomsg
https://blog.csdn.net/u012887259/article/details/102425931