python celery使用队列

在celery的配置方法中有个参数叫task_routes，是用来设置不同的任务 消费不同的队列（也就是路由）。

格式如下：

{ ‘task name’: { ‘queue’: ‘queue name’ }}

直接上代码，简单明了，目录格式如下：

首先是配置文件 config.init.py

import os
import sys
from pathlib import Path

BASE_DIR = Path(__file__).resolve().parent.parent
sys.path.append(str(BASE_DIR))


class Config(object):
    """配置文件基类"""

    """ 项目名称 """
    PROJECT_NAME = "crawler_worker"
    """ celery backend存放结果 """
    CELERY_BACKEND_URL = "redis://127.0.0.1:6379/4"
    """ celery broker中间件 """
    CELERY_BROKER_URL = "redis://127.0.0.1:6379/5"

    """ worker 名称 """
    CRAWL_SEND_EMAIL_TASK = "crawl_service.crawl.send_email_task"  # 抓取发送邮件任务
    CRAWL_SEND_MSG_TASK = "crawl_service.crawl.send_msg_task"  # 抓取发送短信任务


settings = Config()

celery应用程序模块配置相关 celery_base.celery_app.py

import os
import sys
import time
import celery
from pathlib import Path

BASE_DIR = Path(__file__).resolve().parent.parent
sys.path.append(str(BASE_DIR))

from config import settings


# 实例化celery对象
celery_app = celery.Celery(
    settings.PROJECT_NAME,
    backend=settings.CELERY_BACKEND_URL,
    broker=settings.CELERY_BROKER_URL,
    include=[
        "tasks.crawl_send_email",
        "tasks.crawl_send_msg",
    ],
)

# 任务路由
task_routes = {
    settings.CRAWL_SEND_EMAIL_TASK: {
        "queue": f"{settings.CRAWL_SEND_EMAIL_TASK}_queue"
    },
    settings.CRAWL_SEND_MSG_TASK: {"queue": f"{settings.CRAWL_SEND_MSG_TASK}_queue"},
}
# 任务去重
celery_once = {
    "backend": "celery_once.backends.Redis",
    "settings": {"url": settings.CELERY_BACKEND_URL, "default_timeout": 60 * 60},
}
# 配置文件
celery_app.conf.update(
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    task_default_queue="normal",
    timezone="Asia/Shanghai",
    enable_utc=False,
    task_routes=task_routes,
    task_ignore_result=True,
    redis_max_connections=100,
    result_expires=3600,
    ONCE=celery_once,
)

抓取基类 crawl_worker_base.py

from celery_once import QueueOnce


class CrawlBase(QueueOnce):
    """
    抓取worker基类
    """

    name = None
    once = {"graceful": True}
    ignore_result = True

发送邮件任务 crawl_send_email.py

import os
import sys
import time
import celery
from loguru import logger
from pathlib import Path

BASE_DIR = Path(__file__).resolve().parent.parent
sys.path.append(str(BASE_DIR))

from config import settings
from celery_base.celery_app import celery_app
from tasks.crawl_worker_base import CrawlBase

"""

执行命令:
celery -A tasks.crawl_send_email worker -l info -Q crawl_service.crawl.send_email_task_queue

"""


class SendEmailClass(CrawlBase):
    name = settings.CRAWL_SEND_EMAIL_TASK

    def __init__(self, *args, **kwargs):
        super(SendEmailClass, self).__init__(*args, **kwargs)

    def run(self, name):
        logger.info("class的方式, 向%s发送邮件..." % name)
        time.sleep(5)
        logger.info("class的方式, 向%s发送邮件完成" % name)
        return f"成功拿到{name}发送的邮件!"


send_email = celery_app.register_task(SendEmailClass())

发送短信 crawl_send_msg.py

import os
import sys
import time
import celery
from loguru import logger
from pathlib import Path

BASE_DIR = Path(__file__).resolve().parent.parent
sys.path.append(str(BASE_DIR))
from config import settings
from celery_base.celery_app import celery_app
from tasks.crawl_worker_base import CrawlBase

"""

执行命令:
celery -A tasks.crawl_send_msg worker -l info -Q crawl_service.crawl.send_msg_task_queue

"""


class SendMsgClass(CrawlBase):
    name = settings.CRAWL_SEND_MSG_TASK

    def __init__(self, *args, **kwargs):
        super(SendMsgClass, self).__init__(*args, **kwargs)

    def run(self, name):
        logger.info("class的方式, 向%s发送短信..." % name)
        time.sleep(5)
        logger.info("class的方式, 向%s发送短信完成" % name)
        return f"成功拿到{name}发送的短信!"


send_msg = celery_app.register_task(SendMsgClass())

发送邮件任务-调度器 send_email_scheduler.py

import sys
from pathlib import Path

BASE_DIR = Path(__file__).resolve().parent.parent
sys.path.append(str(BASE_DIR))

from config import settings
from celery_base.celery_app import celery_app

if __name__ == "__main__":
    for i in range(100):
        result = celery_app.send_task(
            name=settings.CRAWL_SEND_EMAIL_TASK, args=(f"张三嘿嘿{i}",)
        )
        print(result.id)

发送短信任务-调度器 send_msg_scheduler.py

import os
import sys
import time
from pathlib import Path

BASE_DIR = Path(__file__).resolve().parent.parent
sys.path.append(str(BASE_DIR))

from config import settings
from celery_base.celery_app import celery_app

if __name__ == "__main__":
    for i in range(100, 500):
        result = celery_app.send_task(
            name=settings.CRAWL_SEND_MSG_TASK, args=(f"李四哈哈哈{i}",)
        )
        print(result.id)

准备工作已经做好，紧接着分别执行命令：

celery -A tasks.crawl_send_email worker -l info -Q crawl_service.crawl.send_email_task_queue

celery -A tasks.crawl_send_msg worker -l info -Q crawl_service.crawl.send_msg_task_queue

出现👇🏻下面效果就代表celery启动成功：

最后只要发送任务即可，在redis中就可以看到专门指定的两个队列了。

看下运行过程中的日志

一个简单的celery + 队列就实现了。