Flask项目中Celery的使用

一、项目目录结构

二、celery_config.py

在config中新增文件celery_config.py

# -*- coding: utf8 -*-


# package import
from .redis_config import REDIS_HOST, REDIS_PORT, REDIS_PASSWORD

BROKER_URL = redis://:{}@{}:{}/1.format(REDIS_PASSWORD, REDIS_HOST, REDIS_PORT)
RESULT_BACKEND = redis://:{}@{}:{}/2.format(REDIS_PASSWORD, REDIS_HOST, REDIS_PORT)
TASK_SERIALIZER = json
RESULT_SERIALIZER = json
ACCEPT_CONTENT = [json]
TIMEZONE = Asia/Shanghai
ENABLE_UTC = True

三、task.py

在app.py的同级目录下创建task.py文件

-*- coding: utf8 -*-

# package import
import time
from celery import Celery
from flask import Flask
from config import celery_config

"""
新初始化一个Flask，否则会造成循环引用。
名字一定不能与app.py中的Flask应用重名
例如app.py中的Flask应用的名称为app，tasks.py文件中的Flask应用名称为application
"""
application = Flask(__name__)

application.config.from_object(celery_config)


def make_celery(app):  # 官方提供的使用方法
    celery_server = Celery(app.import_name, backend=app.config[RESULT_BACKEND],
                    broker=app.config[BROKER_URL])
    celery_server.conf.update(app.config)

    task_base = celery_server.Task

    class ContextTask(task_base):
        abstract = True

        def __call__(self, *args, **kwargs):
            with app.app_context():
                return task_base.__call__(self, *args, **kwargs)

    celery_server.Task = ContextTask
    return celery_server


celery = make_celery(application)


# 增加一个实例任务，add()
@celery.task
def add(x, y):

    time.sleep(10)

    return x + y

四、在蓝图中使用

@user_semantic_analysis.route(/hello/celery, methods=[GET, POST])
def hello_celery():
	
	# 使用add()方法
    add.delay(1, 5320)
    return hello celery

五、启动celery服务

# 在终端中输入命令启动celery服务
celery -A tasks worker -l info -P eventlet  -c 10

celery成功调用