django定时器使用 djangocelery定时任务实战详解定时器js代码

一、celery依赖安装

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

二、celery 参数配置

django项目的settings.py中新增如下celery配置

INSTALLED_APPS = [ …… ‘firewall_app’, ‘django_celery_beat’, 添加Celery应用 ‘django_celery_results’, 添加Celery结局展示应用] Celery Configuration Options 使用 Redis 作为消息代理CELERY_BROKER_URL = ‘redis://localhost:6379/0’ 或 ‘amqp://guest:guest@localhost:5672//’ 如果使用 RabbitMQCELERY_RESULT_BACKEND = ‘django-db’CELERY_ACCEPT_CONTENT = [‘json’]CELERY_TASK_SERIALIZER = ‘json’CELERY_RESULT_SERIALIZER = ‘json’CELERY_TIMEZONE = ‘Asia/Shanghai’ 设置时区CELERY_ENABLE_UTC = True Celery Beat Settings (如果使用定时任务)CELERY_BEAT_SCHEDULER = ‘django_celery_beat.schedulers:DatabaseScheduler’ 如果希望在 Django Admin 中管理定时任务，需要安装 django-celery-beat 或者使用默认的本地调度器： 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’: 匹配无论兄弟们的任务模块 ‘handlers’: [‘file’], ‘level’: ‘INFO’, ‘propagate’: True, }, },}

三、celery定时任务

1、新建tasks.py

import loggingimport timefrom celery import shared_tasklogger = logging.getLogger(__name__)def run_crawler_logic(): print(“执行爬虫任务…”) 在这里调用 FortinetCrawler 或相关爬虫函数 crawler = FortinetCrawler() crawler.run() time.sleep(10) 模拟任务执行 print(“爬虫任务完成.”)def run_mapping_logic(): print(“执行漏洞映射任务…”) 在这里调用 map_vulnerabilities_for_all_firewalls 或相关函数 map_vulnerabilities_for_all_firewalls() 推迟导入爬虫函数，避免循环引用 time.sleep(5) 模拟任务执行 print(“漏洞映射任务完成.”)@shared_taskdef run_crawler_task(): “””Celery task for running the web crawler.””” 确保 Django 环境已设置 (如果任务需要访问 Django 模型) 如果 Celery worker 和 Django 运行在同一环境，通常不需要手动设置但为了保险起见，可以加上 if not django.apps.apps.ready: os.environ.setdefault(‘DJANGO_SETTINGS_MODULE’, ‘firewall_monitor.settings’) django.setup() logger.info(“漏洞爬虫任务开始执行”) run_crawler_logic() logger.info(“漏洞爬虫任务完成”) return “漏洞爬虫任务成功执行”@shared_taskdef run_firewall_mapping_task(): “””Celery task for running the firewall vulnerability mapping.””” 同上，确保 Django 环境 if not django.apps.apps.ready: os.environ.setdefault(‘DJANGO_SETTINGS_MODULE’, ‘firewall_monitor.settings’) django.setup() logger.info(“防火墙漏洞映射任务开始执行”) run_mapping_logic() logger.info(“防火墙漏洞映射任务完成”) return “防火墙漏洞映射任务成功执行”

2、新建celery.py

import osfrom celery import Celeryos.environ.setdefault(“DJANGO_SETTINGS_MODULE”, “mysite.settings”)app = Celery(‘firewall_monitor’) 使用 Django settings.py 中的配置app.config_from_object(‘django.conf:settings’, namespace=’CELERY’) 自动发现应用中的任务(最好指定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任务启动

1、Redis-x64-5.0.14.1 window版本 redis启动redis-server.exe redis.windows.conf 2、启动djiango项目python manage.py runserver 8000 3、celery beat启动celery -A mysite beat -l info –scheduler django_celery_beat.schedulers:DatabaseScheduler 4、celery 任务启动celery -A mysite worker –pool=solo -l info -P eventlet

五、celery任务配置

启动djiango项目后，在Django-Admin后台配置定时任务。

六、celery项目结构

到此这篇关于django celery定时任务的文章就介绍到这了,更多相关django celery定时任务内容请搜索风君子博客以前的文章或继续浏览下面的相关文章希望大家以后多多支持风君子博客！

无论兄弟们可能感兴趣的文章:

怎样使用celery进行异步处理和定时任务(django)
django-celery-beat搭建定时任务的实现
关于Django使用 django-celery-beat动态添加定时任务的技巧
Django+Celery实现动态配置定时任务的技巧示例
celery4+django2定时任务的实现代码
django+xadmin+djcelery实现后台管理定时任务
Django配置celery（非djcelery）执行异步任务和定时任务