Python分布式异步任务队列库-Celery学习-02

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
 
from celery import Celery
 
app = Celery(
    "celery_app_name",
    # include中包含的是任务所在的文件名
    include=["celery_worker_tasks"]
)
 
# config_from_object顾名思义，从指定的文件中加载设置
app.config_from_object("celery_project_settings")
 
if __name__ == "__main__":
    pass

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
 
from kombu import Queue
import re
 
 
# 使用redis 作为消息代理
# BROKER_URL 在4.0中变更为 broker_url
# BROKER_URL = "redis://127.0.0.1:6379/0"
broker_url = "redis://:@127.0.0.1:6379/0"
 
# 任务结果存在Redis
# CELERY_RESULT_BACKEND 在4.0中变更为 result_backend
# result_backend = "redis://127.0.0.1:6379/0"
result_backend = "redis://:@127.0.0.1:6379/0"
 
# 任务信息传递的类型(默认json)
# CELERY_RESULT_SERIALIZER 在4.0中变更为 result_serializer
result_serializer = "json"
# CELERY_RESULT_SERIALIZER = "json"
 
# 时区
timezone = "UTC"
 
# 任务过期秒数
# CELERY_TASK_RESULT_EXPIRES 在4.0中变更为 result_expires
# result_expires = 60 * 10
result_expires = 60 * 10
 
# 任务完成指定数量后自动重启
worker_max_tasks_per_child = 1000
 
# List of modules to import when the Celery worker starts.
imports = ('celery_worker_tasks',)
 
# CELERY_QUEUES 在4.0中变更为 task_queues
task_queues = (
    # 定义任务队列
    # 路由键以“default_task.”开头的消息都进default队列
    Queue("default", routing_key="default_task.#"),
    # 路由键以“low_priority_task.”开头的消息都进low_priority_task队列
    Queue("low_priority_task", routing_key="low_priority_task.#"),
    # 路由键以“middle_priority_task.”开头的消息都进middle_priority_task队列
    Queue("middle_priority_task", routing_key="middle_priority_task.#"),
    # 路由键以“higher_priority_task.”开头的消息都进higher_priority_task队列
    Queue("higher_priority_task", routing_key="higher_priority_task.#"),
)
 
# 设置默认队列名为 default
# CELERY_TASK_DEFAULT_QUEUE 在4.0中变更为 task_default_queue
task_default_queue = "default"
# CELERY_TASK_DEFAULT_EXCHANGE 在4.0中变更为 task_default_exchange
task_default_exchange = "default"
# CELERY_TASK_DEFAULT_EXCHANGE_TYPE 在4.0中变更为 task_default_exchange_type
task_default_exchange_type = "direct"
# CELERY_TASK_DEFAULT_ROUTING_KEY 在4.0中变更为 task_default_routing_key
task_default_routing_key = "default_task.default"
 
# CELERY_ROUTES 在4.0中变更为 task_routes
task_routes = (
    [
        (
            #re.compile(r"low"),
            "celery_app_name.celery_worker_tasks.low_rand",
            {"queue": "low_priority_task", "routing_key": "low_priority_task.rand"},
        ),  # 将celery_worker_tasks模块中的low_rand 分配至队列 low_priority_task
        (
            re.compile(r"middle"),
            {"queue": "middle_priority_task", "routing_key": "low_priority_task.rand"},
        ),  # 队列匹配支持正则表达式
    ],
)

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
 
from celery_project_app import app
 
import time
import random
 
 
@app.task
def low_rand(range_start, range_end):
    # 模拟耗时操作
    time.sleep(30)
    return random.randint(range_start, range_end)
 
 
@app.task
def middle_rand(range_start, range_end):
    # 模拟耗时操作
    time.sleep(10)
    return random.randint(range_start, range_end)
 
 
@app.task
def higher_rand(range_start, range_end):
    return random.randint(range_start, range_end)
 
 
if __name__ == "__main__":
    app.worker_main()

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
 
 
import time
# 导入分布任务函数
from celery_worker_tasks import *
 
 
def process_task():
    task_result = 0
    task_object = None
    try:
        # 创建任务
        # 设置超过60秒任务会超时
        task_object = low_rand.apply_async((1,1000), expires=60)
        # 等待任务结束
        while not task_object.ready():
            time.sleep(1)
        if task_object.successful() == True:
            #获取任务结果
            task_result = task_object.get()
    except Exception as e:
        print('celery task unknow except.')
        print(e)
    finally:
        #删除任务和结果
        if task_object:
            task_object.forget()
        print('task return: %s' % task_result)
        pass
 
 
def main():
    while True:
        process_task()
    pass
 
 
if __name__ == '__main__':
    main()
    pass