快捷導(dǎo)航

django celery定時(shí)任務(wù)實(shí)戰(zhàn)詳解

更新時(shí)間：2025年07月01日 10:53:59 作者：cleargy

本文介紹了在Django項(xiàng)目中配置Celery定時(shí)任務(wù)的步驟,包括安裝、參數(shù)設(shè)置、創(chuàng)建任務(wù)文件、啟動(dòng)任務(wù)及后臺(tái)配置,最后整理項(xiàng)目結(jié)構(gòu),本文給大家介紹的非常詳細(xì),感興趣的朋友一起看看吧

一、celery依賴(lài)安裝

# python 3.11版本
pip install celery redis django-celery-beat django_celery_results eventlet

二、celery 參數(shù)配置

django項(xiàng)目的settings.py中新增如下celery配置

##
INSTALLED_APPS = [
    ......
    'firewall_app',
    'django_celery_beat',    # 添加Celery應(yīng)用
    'django_celery_results', # 添加Celery結(jié)果展示應(yīng)用
]
# Celery Configuration Options
# 使用 Redis 作為消息代理
CELERY_BROKER_URL = 'redis://localhost:6379/0'  # 或 'amqp://guest:guest@localhost:5672//' 如果使用 RabbitMQ
CELERY_RESULT_BACKEND = 'django-db'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'Asia/Shanghai' # 設(shè)置時(shí)區(qū)
CELERY_ENABLE_UTC = True
# Celery Beat Settings (如果使用定時(shí)任務(wù))
CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler' # 如果希望在 Django Admin 中管理定時(shí)任務(wù)，需要安裝 django-celery-beat
# 或者使用默認(rèn)的本地調(diào)度器：
# CELERY_BEAT_SCHEDULER = 'celery.beat:PersistentScheduler'
CELERY_BROKER_CONNECTION_RETRY_ON_STARTUP = True
#日志輸出配置
LOGGING = {
    'version': 1,
    'disable_existing_loggers': False,
    'handlers': {
        'file': {
            'level': 'INFO',
            'class': 'logging.FileHandler',
            'filename': 'celery.log',  # 日志文件路徑
        },
    },
    'loggers': {
        'firewall_app.tasks': {  # 匹配您的任務(wù)模塊
            'handlers': ['file'],
            'level': 'INFO',
            'propagate': True,
        },
    },
}

三、celery定時(shí)任務(wù)

1、新建tasks.py

import logging
import time
from celery import shared_task
logger = logging.getLogger(__name__)
def run_crawler_logic():
    print("執(zhí)行爬蟲(chóng)任務(wù)...")
    # 在這里調(diào)用 FortinetCrawler 或相關(guān)爬蟲(chóng)函數(shù)
    # crawler = FortinetCrawler()
    # crawler.run()
    time.sleep(10) # 模擬任務(wù)執(zhí)行
    print("爬蟲(chóng)任務(wù)完成.")
def run_mapping_logic():
    print("執(zhí)行漏洞映射任務(wù)...")
    # 在這里調(diào)用 map_vulnerabilities_for_all_firewalls 或相關(guān)函數(shù)
    # map_vulnerabilities_for_all_firewalls()
    # 推遲導(dǎo)入爬蟲(chóng)函數(shù)，避免循環(huán)引用
    time.sleep(5) # 模擬任務(wù)執(zhí)行
    print("漏洞映射任務(wù)完成.")
@shared_task
def run_crawler_task():
    """Celery task for running the web crawler."""
    # 確保 Django 環(huán)境已設(shè)置 (如果任務(wù)需要訪問(wèn) Django 模型)
    # 如果 Celery worker 和 Django 運(yùn)行在同一環(huán)境，通常不需要手動(dòng)設(shè)置
    # 但為了保險(xiǎn)起見(jiàn)，可以加上
    # if not django.apps.apps.ready:
    #     os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'firewall_monitor.settings')
    #     django.setup()
    logger.info("漏洞爬蟲(chóng)任務(wù)開(kāi)始執(zhí)行")
    run_crawler_logic()
    logger.info("漏洞爬蟲(chóng)任務(wù)完成")
    return "漏洞爬蟲(chóng)任務(wù)成功執(zhí)行"
@shared_task
def run_firewall_mapping_task():
    """Celery task for running the firewall vulnerability mapping."""
    # 同上，確保 Django 環(huán)境
    # if not django.apps.apps.ready:
    #     os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'firewall_monitor.settings')
    #     django.setup()
    logger.info("防火墻漏洞映射任務(wù)開(kāi)始執(zhí)行")
    run_mapping_logic() 
    logger.info("防火墻漏洞映射任務(wù)完成")
    return "防火墻漏洞映射任務(wù)成功執(zhí)行"

2、新建celery.py

import os
from celery import Celery
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "mysite.settings")
app = Celery('firewall_monitor')
# 使用 Django settings.py 中的配置
app.config_from_object('django.conf:settings', namespace='CELERY')
# 自動(dòng)發(fā)現(xiàn)應(yīng)用中的任務(wù)(最好指定tasks路徑)
app.autodiscover_tasks(['mysite.tasks'])
@app.task(bind=True)
def debug_task(self):
    print(f'Request: {self.request!r}')

（3） __init__.py添加

from .celery import app as celery_app
__all__ = ('celery_app',)

四、celery任務(wù)啟動(dòng)

# 1、Redis-x64-5.0.14.1 window版本
# redis啟動(dòng)
redis-server.exe redis.windows.conf
# 2、啟動(dòng)djiango項(xiàng)目
python manage.py runserver 8000
# 3、celery beat啟動(dòng)
celery -A mysite beat -l info --scheduler django_celery_beat.schedulers:DatabaseScheduler
# 4、celery 任務(wù)啟動(dòng)
celery -A mysite worker --pool=solo -l info  -P eventlet