Django与Celery实现异步对列

$ pip install celery==3.1.18

$ pip freeze > requirements.txt

view plain print ?

from __future__ import absolute_import

import os

from celery import Celery

from django.conf import settings

# set the default Django settings module for the 'celery' program.

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'picha.settings')

app = Celery('picha')

# Using a string here means the worker will not have to

# pickle the object when using Windows.

app.config_from_object('django.conf:settings')

app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)

@app.task(bind=True)

def debug_task(self):

print('Request: {0!r}'.format(self.request))

view plain print ?

from __future__ import absolute_import

# This will make sure the app is always imported when

# Django starts so that shared_task will use this app.

from .celery import app as celery_app

├── manage.py

├── picha

│ ├── __init__.py

│ ├── celery.py

│ ├── settings.py

│ ├── urls.py

│ └── wsgi.py

└── requirements.txt

view plain print ?

# CELERY STUFF

BROKER_URL = 'redis://localhost:6379'

CELERY_RESULT_BACKEND = 'redis://localhost:6379'

CELERY_ACCEPT_CONTENT = ['application/json']

CELERY_TASK_SERIALIZER = 'json'

CELERY_RESULT_SERIALIZER = 'json'

CELERY_TIMEZONE = 'Africa/Nairobi'

$ pip install redis==2.10.3

$ pip freeze > requirements.txt

view plain print ?

$ celery -A picha worker -l info

...

[2015-07-07 14:07:07,398: INFO/MainProcess] Connected to redis://localhost:6379//

[2015-07-07 14:07:07,410: INFO/MainProcess] mingle: searching for neighbors

[2015-07-07 14:07:08,419: INFO/MainProcess] mingle: all alone

$ celery -A picha beat -l info

...

[2015-07-07 14:08:23,054: INFO/MainProcess] beat: Starting...

view plain print ?

def add(x, y):

return x + y

view plain print ?

from celery.decorators import task

@task(name="sum_two_numbers")

def add(x, y):

return x + y

├── feedback

│ ├── __init__.py

│ ├── admin.py

│ ├── emails.py

│ ├── forms.py

│ ├── models.py

│ ├── tests.py

│ └── views.py

├── manage.py

├── picha

│ ├── __init__.py

│ ├── celery.py

│ ├── settings.py

│ ├── urls.py

│ └── wsgi.py

├── requirements.txt

└── templates

├── base.html

└── feedback

├── contact.html

└── email

├── feedback_email_body.txt

└── feedback_email_subject.txt

view plain print ?

from celery.decorators import task

from celery.utils.log import get_task_logger

from feedback.emails import send_feedback_email

logger = get_task_logger(__name__)

@task(name="send_feedback_email_task")

def send_feedback_email_task(email, message):

"""sends an email when feedback form is filled successfully"""

logger.info("Sent feedback email")

return send_feedback_email(email, message)

view plain print ?

from django import forms

from feedback.tasks import send_feedback_email_task

class FeedbackForm(forms.Form):

email = forms.EmailField(label="Email Address")

message = forms.CharField(

label="Message", widget=forms.Textarea(attrs={'rows': 5}))

honeypot = forms.CharField(widget=forms.HiddenInput(), required=False)

def send_email(self):

# try to trick spammers by checking whether the honeypot field is

# filled in; not super complicated/effective but it works

if self.cleaned_data['honeypot']:

return False

send_feedback_email_task.delay(

self.cleaned_data['email'], self.cleaned_data['message'])

view plain print ?

from celery.task.schedules import crontab

from celery.decorators import periodic_task

@periodic_task(run_every=(crontab(minute='*/15')), name="some_task", ignore_result=True)

def some_task():

# do something

├── feedback

│ ├── __init__.py

│ ├── admin.py

│ ├── emails.py

│ ├── forms.py

│ ├── models.py

│ ├── tasks.py

│ ├── tests.py

│ └── views.py

├── manage.py

├── photos

│ ├── __init__.py

│ ├── admin.py

│ ├── models.py

│ ├── settings.py

│ ├── tests.py

│ ├── utils.py

│ └── views.py

├── picha

│ ├── __init__.py

│ ├── celery.py

│ ├── settings.py

│ ├── urls.py

│ └── wsgi.py

├── requirements.txt

└── templates

├── base.html

├── feedback

│ ├── contact.html

│ └── email

│ ├── feedback_email_body.txt

│ └── feedback_email_subject.txt

└── photos

└── photo_list.html

view plain print ?

from celery.task.schedules import crontab

from celery.decorators import periodic_task

from celery.utils.log import get_task_logger

from photos.utils import save_latest_flickr_image

logger = get_task_logger(__name__)

@periodic_task(

run_every=(crontab(minute='*/15')),

name="task_save_latest_flickr_image",

ignore_result=True

)

def task_save_latest_flickr_image():

"""

Saves latest image from Flickr

"""

save_latest_flickr_image()

logger.info("Saved image from Flickr")

$ celery -A picha worker -l info

$ celery -A picha beat -l info

view plain print ?

; ==================================

; celery worker supervisor example

; ==================================

; the name of your supervisord program

[program:pichacelery]

; Set full path to celery program if using virtualenv

command=/home/mosh/.virtualenvs/picha/bin/celery worker -A picha --loglevel=INFO

; The directory to your Django project

directory=/home/mosh/sites/picha

; If supervisord is run as the root user, switch users to this UNIX user account

; before doing any processing.

user=mosh

; Supervisor will start as many instances of this program as named by numprocs

numprocs=1

; Put process stdout output in this file

stdout_logfile=/var/log/celery/picha_worker.log

; Put process stderr output in this file

stderr_logfile=/var/log/celery/picha_worker.log

; If true, this program will start automatically when supervisord is started

autostart=true

; May be one of false, unexpected, or true. If false, the process will never

; be autorestarted. If unexpected, the process will be restart when the program

; exits with an exit code that is not one of the exit codes associated with this

; process’ configuration (see exitcodes). If true, the process will be

; unconditionally restarted when it exits, without regard to its exit code.

autorestart=true

; The total number of seconds which the program needs to stay running after

; a startup to consider the start successful.

startsecs=10

; Need to wait for currently executing tasks to finish at shutdown.

; Increase this if you have very long running tasks.

stopwaitsecs = 600

; When resorting to send SIGKILL to the program to terminate it

; send SIGKILL to its whole process group instead,

; taking care of its children as well.

killasgroup=true

; if your broker is supervised, set its priority higher

; so it starts first

priority=998

view plain print ?

; ================================

; celery beat supervisor example

; ================================

; the name of your supervisord program

[program:pichacelerybeat]

; Set full path to celery program if using virtualenv

command=/home/mosh/.virtualenvs/picha/bin/celerybeat -A picha --loglevel=INFO

; The directory to your Django project

directory=/home/mosh/sites/picha

; If supervisord is run as the root user, switch users to this UNIX user account

; before doing any processing.

user=mosh

; Supervisor will start as many instances of this program as named by numprocs

numprocs=1

; Put process stdout output in this file

stdout_logfile=/var/log/celery/picha_beat.log

; Put process stderr output in this file

stderr_logfile=/var/log/celery/picha_beat.log

; If true, this program will start automatically when supervisord is started

autostart=true

; May be one of false, unexpected, or true. If false, the process will never

; be autorestarted. If unexpected, the process will be restart when the program

; exits with an exit code that is not one of the exit codes associated with this

; process’ configuration (see exitcodes). If true, the process will be

; unconditionally restarted when it exits, without regard to its exit code.

autorestart=true

; The total number of seconds which the program needs to stay running after

; a startup to consider the start successful.

startsecs=10

; if your broker is supervised, set its priority higher

; so it starts first

priority=999

$ touch /var/log/celery/picha_worker.log

$ touch /var/log/celery/picha_beat.log

$ sudo supervisorctl reread

$ sudo supervisorctl update

$ sudo supervisorctl stop pichacelery

$ sudo supervisorctl start pichacelery

$ sudo supervisorctl status pichacelery