celery消费任务和任务定时的操作

开发环境

ubuntu 18.04 + python3.6.9 / windows10 + python 3.6.9

安装依赖

# 安装django==2.2.17
pip install django==2.2.17
# 安装celery == 4.4.0
pip install celery==4.4.0
# 安装 redis == 3.5.3
pip install redis==3.5.3
# 安装 flower == 0.9.7
pip install flower==0.9.7

创建 django 项目和应用

#  创建项目(mysite)
django-admin startproject mysite
# 切换到mysite中
cd mysite
# 创建应用 (user)
python manage.py startapp user

在 mysite 目录下新建 celery_task 包

在 celery_task 中新建 celery.py 和 config.py

# config.py
from __future__ import absolute_import  # 拒绝隐式引入，因为celery.py的名字和celery的包名冲突，需要使用这条语句让程序正确地运行
from celery.schedules import crontab
from nest_e100.settings import REDIS_HOST, REDIS_PORT, REDIS_PASSWORD

broker_url = "redis://:{}@{}:{}/98".format(REDIS_PASSWORD, REDIS_HOST, REDIS_PORT)  # 使用redis存储任务队列
result_backend = "redis://:{}@{}:{}/99".format(REDIS_PASSWORD, REDIS_HOST, REDIS_PORT)  # 使用redis存储结果

task_serializer = 'json'
result_serializer = 'json'
accept_content = ['json']
timezone = "Asia/Shanghai"  # 时区设置
worker_hijack_root_logger = False  # celery默认开启自己的日志，可关闭自定义日志，不关闭自定义日志输出为空
result_expires = 60 * 60 * 24  # 存储结果过期时间（默认1天）

# 导入任务所在文件
imports = [
    "celery_task.crontab.company_task",  # 导入py文件
]

# 需要执行任务的配置
beat_schedule = {
    "crontab_company_task": {
        "task": "celery_task.crontab.company_task.check_company_task",  # 执行的函数
        "schedule": crontab(minute="*/1"), # 每分钟执行一次
        "args": ()  # # 任务函数参数
    },
}

# celery.py
from celery import Celery
import django, os, sys
# 获取根目录
base_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "nest_e100")
sys.path.append(base_dir)
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'nest_e100.settings')
# 实例化django项目中的每一个app实例
django.setup()
# 创建celery应用对象
app = Celery("celery_demo")

# 导入celery的配置信息
app.config_from_object("celery_task.config")

新建定时任务和任务消费

# 在 celery_task/crontab/company_task.py 做公司定时任务
from celery_task.celery import app
from .data_access.cache_manager import RedisUtil
from celery_task.tasks.company_task import e100_company_task
from celery_task.common.mytime import cal_count_down_time
from .utils.logger import logger

# 实例化redis
redis_util_cache = RedisUtil("openapi")

# 检查商户详情信息
def check_company_info():
    data = redis_util_cache.get_key_values("company*")
    if not data:
        logger.warning(">>>>>>>>>>>>>>>>>> redis中没有公司信息,请检查...")
        return False
    for i in data:
        key = i.get("key", "")
        if key:
            # 计算该key 剩余的时间的秒数
            time_left_timestamp = redis_util_cache.ttl(key)
            # 计算时间差是否在规定范围内，如果在，返回True,则添加队列中，否则返回False,不做任何处理
            flag = cal_count_down_time(time_left_timestamp)
            # 添加队列中
            if flag: e100_company_task.delay({"com_id": key[8:]})
    print(">>>>>>>> crontab check company detail info is end ..")


@app.task
def check_company_task():
    check_company_info()


if __name__ == '__main__':
    check_company_task()

# 在 celery_task/task/company_task.py 做公司任务消费
from celery_task.celery import app
from .data_access.cache_manager import RedisUtil
from .utils.account_manager import request_to_openplatform_manage
from .utils.external_url_request import external_system_url
from django.conf import settings
import json
# 实例化redis
openapi_redis_cache = RedisUtil("openapi")

# 更新商户详情信息
@app.task
def e100_company_task(data):
    # 请求开发平台获取数据
    response = request_to_openplatform_manage(data, external_system_url.get_openapi_company_details_info)
    # 获取的数据
    ret = response.data.get("data", "")
    # 将数据放到cache中， key:  company_公司Id
    openapi_redis_cache.set("company_%s" % (data.get("com_id")), json.dumps(ret),
                            expire_time=settings.TOKEN_CACHE_TIME)

启动任务

# 启动django项目
python manage.py runserver
# 启动celery 任务, 如果是windows启动，需要在后面添加 --pool=solo
celery -A celery_task.celery worker -l info 
# 启动celery定时
celery -A celery_task.celery beat
# 启动flower
celery -A celery_task flower

访问 http://127.0.0.1:5555
最终的项目结构为: