在Web开发的过程中，难免有一些操作是耗时任务，有时候我们不想阻塞Web进程，希望提交到后台处理，有结果了再通知前端；有时候用户不需要实时结果，可以异步处理的，比如发送邮件通知；有时候我们希望定时执行某些任务，复用已有的一些Web代码。对于第一种情况，可以是RPC或调用其他接口处理；第二种情况则是放入队列，由后台进程异步处理；第三种情况可以是在定时任务内处理，也可以是触发消息放入队列，由后台进程任务处理，同第二种情况。对于第一种情况，也可以是放入队列由后台进程处理，Web前端，定时轮询队列/接口是否有结果。这样三种情况都可以统一为一种情况，即不同来源事件（用户/定时器）触发消息，放入队列由后台任务进程异步处理，异步查询结果。
之前的一个Python项目即类似。不管代码是PHP还是Python思想都是类似，只是不同语言有不同工具/优势来处理。Web前端展示统计报表，后台进程定时查询Impala，分析统计数据；Web前端也可以触发事件比如统计、发送邮件报告等，后台进程监听并处理。Web前端选择Django，是Python写的一个MVT框架，提供了登录认证、管理后台模块，还有命令行助手工具可以生成项目，易于快速搭建并扩展Web站点。后台任务处理选择Celery，是Python写的一个分布式的任务处理平台，支持各种消息队列，比如RabbbitMQ、Redis，并提供Flower监控工具。Django本身是一个Web框架，生产环境最好使用性能更好的HTTP服务器，Gunicorn是Python写的一个WSGI HTTP服务器，用来监听HTTP请求，调用Django。Gunicorn前端最好使用Nginx来做代理，分发请求，结构如下：
安装Python 3和Django，并生成项目框架代码。注意：由于机器上面本来就有Pyton 2.7，所以要重命名下。
yum -y update yum -y install yum-utils yum -y groupinstall development yum install zlib-devel bzip2-devel openssl-devel ncurese-devel expat-devel gdbm-devel readline-devel sqlite-devel libffi-devel wget https://www.python.org/ftp/python/3.6.2/Python-3.6.2.tar.xz tar Jxvf Python-3.6.2.tar.xz mv Python-3.6.2 /opt/Python-3.6.2 cd /opt/Python-3.6.2 ./configure --enable-shared --prefix=/usr/local/python3 make && make install vim /etc/ld.so.conf /usr/local/python3/lib/ /sbin/ldconfig -v ln -s /usr/local/python3/bin/python3.6 /usr/bin/python3.6 ln -s /usr/bin/python3.6 /usr/bin/python3 #ln -s /usr/local/python3/bin/python3.6 /usr/bin/python ln -s /usr/local/python3/bin/pip3 /usr/bin/pip3 python3 -V pip3 -V pip3 install Django python3 -m django --version ln -s /usr/local/python3/bin/django-admin /usr/bin/django-admin/ cd /home/rc django-admin startproject qrd cd qrd python3 manage.py startapp master vim qrd/settings.py
编辑settings.py，监听相应域名、IP的请求
ALLOWED_HOSTS = ['dev.example.com','localhost','127.0.0.1','10.1.*.*]
INSTALLED_APPS = [
'django.contrib.admin',
'django.contrib.auth',
'master',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.messages',
'django.contrib.staticfiles',
]
测试一下
python3 manage.py runserver 0:80
然后安装Gunicorn
$ pip3 install gunicorn $ pip3 install greenlet $ pip3 install ConfigParser $ ln -s /usr/local/python3/bin/gunicorn /usr/bin/gunicorn
测试运行
$ cd ~/qrd/qrd $ gunicorn -w 4 qrd.wsgi --bind unix:/tmp/gunicorn.sock $ sudo vim /etc/systemd/system/gunicorn.service
创建Gunicorn服务
[Unit] Description=gunicorn daemon After=network.target [Service] User=root Group=root WorkingDirectory=/home/rc/qrd/qrd ExecStart=/usr/bin/gunicorn --access-logfile - --workers 3 --bind unix:/tmp/gunicorn.sock qrd.wsgi [Install] WantedBy=multi-user.target
开机启动
systemctl start gunicorn systemctl enable gunicorn #systemctl stop gunicorn
接下来安装Nginx
yum install nginx vim /etc/nginx/conf.d/default.conf
配置Nginx与Gunicorn通过UNIX Sockect通信
server {
listen 80;
server_name localhost;
#charset koi8-r;
#access_log /var/log/nginx/host.access.log main;
#location / {
# root /usr/share/nginx/html;
# index index.html index.htm;
#
location / {
proxy_pass http://unix:/tmp/gunicorn.sock;
proxy_set_header Host $host;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
}
}
启动Nginx
systemctl start nginx systemctl enable nginx
安装MySQL及RabbitMQ
yum install mysql-community-devel pip3 install mysqlclient yum install rabbitmq-server rabbitmq-server -detached rabbitmqctl status
创建RabbitMQ vhost
$ rabbitmqctl add_user qrdweb <password> Creating user "qrdweb" ... ...done. $ rabbitmqctl add_vhost qrdweb Creating vhost "qrdweb" ... ...done. $ rabbitmqctl set_user_tags qrdweb management Setting tags for user "qrdweb" to [management] ... ...done. $ rabbitmqctl set_permissions -p qrdweb qrdweb ".*" ".*" ".*" Setting permissions for user "qrdweb" in vhost "qrdweb" ... ...done. $ netstat -apn | grep rabbitmq $ rabbitmqctl status
安装Celery
pip3 install celery ln -s /usr/local/python3/bin/celery /usr/bin/celery
测试Celery的异步任务worker及计划任务schedule
$ cd /home/qrd/qrd/ $ ls dashboard db.sqlite3 manage.py master qrd static $ celery -A qrd worker -l info /usr/local/python3/lib/python3.6/site-packages/celery/platforms.py:795: RuntimeWarning: You're running the worker with superuser privileges: this is absolutely not recommended! Please specify a different user using the -u option. User information: uid=0 euid=0 gid=0 egid=0 uid=uid, euid=euid, gid=gid, egid=egid, -------------- [email protected] v4.1.0 (latentcall) ---- **** ----- --- * *** * -- Linux-3.10.0-693.2.2.el7.x86_64-x86_64-with-centos-7.4.1708-Core 2017-09-22 08:19:39 -- * - **** --- - ** ---------- [config] - ** ---------- .> app: qrd:0x7fda62e705c0 - ** ---------- .> transport: amqp://qrdweb:**@localhost:5672/qrdweb - ** ---------- .> results: disabled:// - *** --- * --- .> concurrency: 1 (prefork) -- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker) --- ***** ----- -------------- [queues] .> celery exchange=celery(direct) key=celery [tasks] . dashboard.tasks.debug_task [2017-09-22 08:19:39,769: INFO/MainProcess] Connected to amqp://qrdweb:**@127.0.0.1:5672/qrdweb [2017-09-22 08:19:39,781: INFO/MainProcess] mingle: searching for neighbors [2017-09-22 08:19:40,811: INFO/MainProcess] mingle: all alone [2017-09-22 08:19:40,860: WARNING/MainProcess] /usr/local/python3/lib/python3.6/site-packages/celery/fixups/django.py:202: UserWarning: Using settings.DEBUG leads to a memory leak, never use this setting in production environments! warnings.warn('Using settings.DEBUG leads to a memory leak, never ' [2017-09-22 08:19:40,860: INFO/MainProcess] [email protected] ready. [2017-09-22 08:20:55,023: INFO/MainProcess] Received task: dashboard.tasks.debug_task[71e6c0e1-92e1-494e-b5e9-163eeb7bd24e] [2017-09-22 08:20:55,027: INFO/ForkPoolWorker-1] Task dashboard.tasks.debug_task[71e6c0e1-92e1-494e-b5e9-163eeb7bd24e] succeeded in 0.001253978000022471s: 'debug_task' [2017-09-22 08:22:21,179: INFO/MainProcess] Received task: dashboard.tasks.debug_task[b81fe9a0-1725-4702-ba0e-13196c9b5977] [2017-09-22 08:22:21,180: INFO/ForkPoolWorker-1] Task dashboard.tasks.debug_task[b81fe9a0-1725-4702-ba0e-13196c9b5977] succeeded in 0.00018433199147693813s: 'debug_task' $ celery -A qrd beat -l info -s /tmp/celerybeat-schedule celery beat v4.1.0 (latentcall) is starting. __ - ... __ - _ LocalTime -> 2017-09-24 04:20:37 Configuration -> . broker -> amqp://qrdweb:**@localhost:5672/qrdweb . loader -> celery.loaders.app.AppLoader . scheduler -> celery.beat.PersistentScheduler . db -> /tmp/celerybeat-schedule . logfile -> [stderr]@%INFO . maxinterval -> 5.00 minutes (300s) [2017-09-24 04:20:37,823: INFO/MainProcess] beat: Starting... [2017-09-24 04:20:37,866: INFO/MainProcess] Scheduler: Sending due task add every 10 (qrd.celery.test) [2017-09-24 04:20:47,856: INFO/MainProcess] Scheduler: Sending due task add every 10 (qrd.celery.test) [2017-09-24 04:20:57,858: INFO/MainProcess] Scheduler: Sending due task add every 10 (qrd.celery.test) [2017-09-24 04:20:57,861: INFO/MainProcess] Scheduler: Sending due task qrd.celery.test('world') (qrd.celery.test) [2017-09-24 04:21:07,858: INFO/MainProcess] Scheduler: Sending due task add every 10 (qrd.celery.test) [2017-09-24 04:21:17,859: INFO/MainProcess] Scheduler: Sending due task add every 10 (qrd.celery.test)
运行成功，可以使用Supervisord来守护监控Celery的运行，参考这里。
Django项目结果如下
先配置Celery使用RabbitMQ作为Broker，使用Django DB来保存调用结果settings.py
import os
from configparser import RawConfigParser
#https://code.djangoproject.com/wiki/SplitSettings
config = RawConfigParser()
config.read('/home/qrd/setting/settings.ini')
STATIC_URL = '/static/'
STATIC_ROOT = os.path.join(BASE_DIR, 'static')
CELERY_BROKER_URL = 'amqp://usr:pwd@localhost:5672/qrdweb'
CELERY_RESULT_BACKEND = 'django-db'
然后在Django项目下创建celery.py文件，
from __future__ import absolute_import, unicode_literals
import os
from celery import Celery
from celery.schedules import crontab
# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'qrd.settings')
app = Celery('qrd')
# Using a string here means the worker doesn't have to serialize
# the configuration object to child processes.
# - namespace='CELERY' means all celery-related configuration keys
# should have a `CELERY_` prefix.
app.config_from_object('django.conf:settings', namespace='CELERY')
# Load task modules from all registered Django app configs.
app.autodiscover_tasks()
app.conf.beat_schedule = {
'hue-tasks-debug_task': {
'task': 'hue.tasks.debug_task',
'schedule': 10.0,
'args': ()
},
}
并且在__init__.py引入Celery即可集成
from __future__ import absolute_import, unicode_literals # 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 __all__ = ['celery_app']
Django的异步任务只能定义在各个app的task.py文件里，比如qrd.hue.tasks定义了一个定时任务
from celery import task
@task
def debug_task():
#print(arg)
return 'debug_task'
也可以在其模块里面调用
from tasks import debug_task
def save(data):
debug_task.delay()
顺便推荐一个Bootstrap管理后台模板：gentelella。
参考链接：
异步任务神器 Celery
Django配置celery执行异步任务和定时任务
淺談 Gunicorn 各個 worker type 適合的情境