How to resolve the issue of repeated execution of Django scheduled tasks?
In Django, the Celery library can be used to achieve the repeated execution of scheduled tasks. Celery is a distributed task queue that allows you to asynchronously execute tasks in the background, including scheduled tasks.
Here is a simple example of using Celery to implement scheduled tasks.
- To install Celery: First, you need to install the Celery library. You can use pip to install Celery.
pip install celery
- Create a Celery instance: In a Django project, create a celery.py file to configure the Celery instance.
from celery import Celery
import os
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project.settings')
app = Celery('your_project')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()
- Create a task: in your Django application, create a tasks.py file and define the tasks that need to be executed.
from celery import shared_task
@shared_task
def my_task():
# 你的任务逻辑
- To start the Celery worker, run the following command in the project root directory:
celery -A your_project worker --loglevel=info
- Set up scheduled tasks: You can use Celery’s task scheduler to configure the execution time of tasks. For example, add the following code in the celery.py file.
from celery.schedules import crontab
app.conf.beat_schedule = {
'my-task': {
'task': 'your_app.tasks.my_task',
'schedule': crontab(hour=0, minute=0), # 每天凌晨执行任务
},
}
The above is a simple example of implementing scheduled tasks using the Celery library. With Celery’s task scheduler, you can configure the execution time of tasks to have them repeat at set intervals or time points.