Using celery in projects

Project structure

The structure of the project is as follows

celery.py

from celery import Celery

#  establish celery example 
app = Celery('demo')
app.config_from_object('project.celeryconfig')

#  Automatic search task 
app.autodiscover_tasks(['project'])

celeryconfig.py

broker_url = 'redis://127.0.0.1:6379/5'
result_backend = 'redis://127.0.0.1:6379/6'

tasks.py

from project.celery import app as celery_app


#  Create a task function 
@celery_app.task
def my_task1():
    print(" Task function (my_task1) Being implemented ....")


@celery_app.task
def my_task2():
    print(" Task function (my_task2) Being implemented ....")


@celery_app.task
def my_task3():
    print(" Task function (my_task3) Being implemented ....")

Start project

Click on

explain

1. The task searched
2. Configuration information

route

The website links

1. When there are many tasks, we can specify a queue to deal with specific things
2. For example, a queue handles sending mail , A queue processes uploaded files

Start the queue

We start two queues Queue1 and Queue2, Get into Linux System , Note that we need to start two terminals , Input the following commands respectively

celery -A project worker --loglevel=info -Q queue1
celery -A project worker --loglevel=info -Q queue2

• So we started two queues to work , This is equivalent to starting two worker The server , One worker The server is responsible for a queue ( How to use this demo )
• We can also use a worker The server handles two queues , Enter the following command

celery -A project worker --loglevel=info -Q queue1,queue2

Routing configuration

The so-called routing configuration , That is, configure which task to which queue ( What line of work )

1. open celeryconfig.py file , Configure as follows

task_routes=({
    
    'project.tasks.my_task1': {
    'queue': 'queue1'},
    'project.tasks.my_task2': {
    'queue': 'queue1'}
    })

2. Create a new one under the project test.py Write the following code

Scheduling tasks

delay

1. We have another terminal , have access to delay To schedule tasks
2. Enter the following statement

from project.tasks import my_task1
my_task1.delay()

We will find that the task has been executed

apply_async

Use this method to schedule tasks , You can specify the delay time of queue and task execution

my_task2.apply_async((),queue="queue1",countdown=10)

• The current function refers to sending tasks to queue1 queue ,10s After execution

Cycle task

Here we will use celery beat Scheduler , He can make the task cycle , Relevant instructions on the official website , Click on

Normal cycle task

Suppose that every 5S Perform a task 3, Then we can celeryconfig.py Add the following configuration under the file

beat_schedule = {
    
    "every-5-seconds": {
    
        "task": "project.tasks.my_task3",
        "schedule": 5.0
    }
}

start-up Linux System , Enter the command

celery -A project worker --loglevel=info --beat

When we start , You will find that the task starts every 5s Automatic execution

Timed periodic tasks

If we want to perform tasks every day or every week , So this is the time to use cron Mission , Can be in celeryconfig.py Add the following configuration information

from celery.schedules import crontab
beat_schedule = {
    
    "add-every-monday-morning": {
    
        "task": "project.tasks.my_task2",
        "schedule": crontab(hour=7, minute=30, day_of_week=1)
    }
}

So what this means is Every Monday morning 7:30 Perform tasks 3

Mission time

celerybeat-schedule The result of task execution time is saved in celerybeat-schedule In the series