在 上一篇文章 中，我们对 Celery 有了初步的认识，接下来就通过两个例子来进行实战。最终的效果图如下：
项目 github 地址：https://github.com/happy-python/flask_celery
使用环境及版本：
- python 2.7.10
- celery==4.0.0
- Flask==0.11.1
- Flask-Mail==0.9.1
- flower==0.9.1
- redis==2.10.5
简单例子：异步发送邮件
此示例是应用程序非常普通的需求：能够发送邮件但是不阻塞主应用。
本例使用了 Flask-Mail 扩展。
- Flask-Mail==0.9.1
此示例应用是一个只有一个输入文本框的简单表单。要求用户在此文本框中输入一个电子邮件地址，点击提交，服务器会发送一个测试电子邮件到这个邮件地址。表单中包含两个提交按钮，一个立即发送邮件，一个是一分钟后发送邮件。
<h2>异步发送邮件</h2>
{% for message in get_flashed_messages() %}
<p style="color: red;">{{ message }}</p>
{% endfor %}
<form method="POST" action="{{ url_for('index') }}">
<p>Send test email to: <input type="text" name="email" value="{{ email }}"></p>
<input type="submit" name="submit" value="Send">
<input type="submit" name="submit" value="Send in 1 minute">
</form>
这里没有什么特别的东西。只是一个普通的 HTML 表单，再加上 Flask 闪现消息。
Flask-Mail 扩展配置
# Flask-Mail 配置
app.config['MAIL_SERVER'] = 'smtp.163.com' # 电子邮件服务器的主机名或IP地址
app.config['MAIL_PORT'] = 25 # 电子邮件服务器的端口
app.config['MAIL_USE_TLS'] = True # 启用传输层安全协议
app.config['MAIL_USE_SSL'] = False # 启用安全套接层协议
app.config['MAIL_USERNAME'] = 'your-mail-username' # 邮件账户用户名
app.config['MAIL_PASSWORD'] = 'your-mail-password' # 邮件账户的密码
后台路由处理
@app.route('/', methods=['GET', 'POST'])
def index():
if request.method == 'GET':
return render_template('index.html', email=session.get('email', ''))
email = request.form['email']
session['email'] = email
msg = Message('Hello from Flask', sender=app.config['MAIL_USERNAME'], recipients=[email])
msg.body = 'This is a test email sent from a background Celery task.'
if request.form['submit'] == 'Send':
# 立即发送
# delay 是 apply_async 的快捷快捷方式
# 相比于 delay，当使用 apply_async 时，我们能够对后台任务的执行方式有更多的控制。例如任务在何时执行
# delay 和 apply_async 的返回值是一个 AsyncResult 的对象。通过该对象，能够获得任务的状态信息
async_send_email.delay(msg)
flash('Sending email to {0}'.format(email))
else:
# 1分钟后发送
async_send_email.apply_async(args=[msg], countdown=60)
flash('An email will be sent to {0} in one minute'.format(email))
return redirect(url_for('index'))
在这里将用户在文本框中输入的值保存在 session 中，以便页面重新加载时能够使用该信息。
后台异步任务
@celery.task
def async_send_email(msg):
app = create_app()
with app.app_context():
Mail(app).send(msg)
任何需要作为后台任务的函数都需要使用 celery.task 装饰器装饰。
需要注意的是 Flask-Mail 需要在应用的上下文中运行，因此在调用 send() 之前需要创建一个应用上下文。
重点注意此异步调用返回值并不保留，因此应用本身无法知道是否调用成功或者失败。运行这个示例的时候，需要检查 Celery worker 的输出来排查发送邮件过程是否有问题。
复杂例子：显示进度更新和结果
此示例展示一个虚构的长时间运行的任务。用户点击按钮启动一个或者更多的长时间运行的任务，在浏览器上的页面使用 ajax 轮询服务器更新所有任务的状态。每一个任务，页面都会显示一个进度条，一个当前进度信息和一个当前执行结果。
后台异步任务
@celery.task(bind=True)
def long_task(self):
total = random.randint(10, 50)
for i in range(total):
# 自定义状态 state
self.update_state(state=u'处理中', meta={'current': i, 'total': total})
time.sleep(1)
return {'current': 100, 'total': 100, 'result': u'完成'}
对于这个任务（此任务在一个 Celery worker 进程中运行），在 Celery 装饰器中添加了 bind=True 参数，这使得 Celery 向函数中传入了 self 参数，因此在函数中能够使用它(self)来记录状态更新。
self.update_state() 指明 Celery 如何接收任务更新。有一些内置的状态，比如 STARTED，SUCCESS 等等，但是 Celery 也支持自定义状态。在这里使用了一个叫做 '处理中' 的自定义状态。连同状态，还有一个元数据，该元数据是 Python 字典形式，包含目前和总的迭代数。客户端可以使用这些元素来显示一个漂亮的进度条。每迭代一次休眠一秒，以模拟正在做一些工作。当循环退出，一个 Python 字典作为函数结果返回。
启动后台任务
@app.route('/longtask')
def longtask():
# 开启异步任务
task = long_task.apply_async()
return jsonify({}), 202, {'Location': url_for('taskstatus', task_id=task.id)}
客户端需要发起一个 GET 请求到 /longtask 来触发后台任务执行。对于响应状态码 202，这个状态码通常是在 REST API 中使用用来表明一个请求正在处理中。同时添加了 Location 头，值为一个客户端用来获取状态信息的 URL。这个 URL 指向另一个叫做 taskprogress 的 Flask 路由，并且该 URL 包含 task.id。
获取任务状态信息的路由
@app.route('/status/<task_id>')
def taskstatus(task_id):
# 获取异步任务结果
task = long_task.AsyncResult(task_id)
# 等待处理
if task.state == 'PENDING':
response = {'state': task.state, 'current': 0, 'total': 1}
elif task.state != 'FAILURE':
response = {'state': task.state, 'current': task.info.get('current', 0), 'total': task.info.get('total', 1)}
# 处理完成
if 'result' in task.info:
response['result'] = task.info['result']
else:
# 后台任务出错
response = {'state': task.state, 'current': 1, 'total': 1}
return jsonify(response)
这个路由生成一个 JSON 响应，该响应包含任务的状态以及在 update_state() 调用中设置的 meta 参数的所有值。
第一个 if 代码块是当任务还没有开始的时候(PENDING 状态)。在这种情况下暂时没有状态信息，因此人为地制造了些数据。接下来的 elif 代码块返回后台任务的状态信息。任务提供的信息可以通过访问 task.info 获得。如果数据中包含 result ，这就意味着这是最终的结果并且任务已经结束，因此我把这些信息也加到响应中。最后的 else 代码块是任务执行失败的情况，在这种情况下 task.info 中会包含异常的信息。
客户端
<h2>显示进度更新和结果</h2>
<button id="start-bg-job">Start Long Task</button>
<br><br>
<div id="progress" style="width: auto;text-align: center;"></div>
对于图形进度条使用 nanobar.js。同样还需要引入 jQuery，它能够简化 ajax 的调用。
<script src="//cdn.bootcss.com/nanobar/0.4.2/nanobar.min.js"></script>
<script src="//cdn.bootcss.com/jquery/3.1.1/jquery.min.js"></script>
启动后台任务：
$('button').on('click', start_long_task);
function start_long_task() {
// 添加任务状态元素
var div = $('<div class="progress"><div></div><div>0%</div><div></div></div><hr>');
$('#progress').append(div);
// 创建进度条（progress bar）
var nanobar = new Nanobar({
bg: '#44f',
target: div[0].childNodes[0]
});
// 向后台发送请求开启任务
var longTask = $.get('/longtask');
longTask.done(function (data, status, request) {
status_url = request.getResponseHeader('Location');
update_progress(status_url, nanobar, div[0]);
});
}
div 代码解释：
<div class="progress">
<div></div> <-- 进度条
<div>0%</div> <-- 当前进度
<div> </div> <-- 当前结果
</div>
<hr>
更新进度信息：
function update_progress(status_url, nanobar, status_div) {
// 获取状态信息
$.getJSON(status_url, function (data) {
// 更新进度
percent = parseInt(data['current'] * 100 / data['total']);
nanobar.go(percent);
$(status_div.childNodes[1]).text('当前进度: ' + percent + '%');
// 轮询
if (data['state'] == 'PENDING' || data['state'] == '处理中') {
setTimeout(function () {
update_progress(status_url, nanobar, status_div);
}, 2000);
}
// 更新结果
if ('result' in data) {
// 处理完成
$(status_div.childNodes[2]).text('当前结果: ' + data['result']);
}
else {
// 处理中
$(status_div.childNodes[2]).text('当前结果: ' + data['state']);
}
});
}
当后台任务运行时，为了能够持续获得任务状态并更新页面，这里使用了定时器，每隔两秒更新一次直到后台任务完成。
运行
启动 celery worker：
$ celery -A app worker -l info
-------------- celery@ubuntu v4.0.0 (latentcall)
---- **** -----
--- * *** * -- Linux-3.16.0-57-generic-i686-with-Ubuntu-14.04-trusty 2016-11-26 22:05:09
-- * - **** ---
- ** ---------- [config]
- ** ---------- .> app: app:0xb6472acc
- ** ---------- .> transport: redis://localhost:6379/15
- ** ---------- .> results: redis://localhost:6379/15
- *** --- * --- .> concurrency: 4 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** -----
-------------- [queues]
.> celery exchange=celery(direct) key=celery
[tasks]
. app.tasks.async_send_email
. app.tasks.long_task
[2016-11-26 22:05:10,177: INFO/MainProcess] Connected to redis://localhost:6379/15
[2016-11-26 22:05:10,192: INFO/MainProcess] mingle: searching for neighbors
[2016-11-26 22:05:11,231: INFO/MainProcess] mingle: all alone
[2016-11-26 22:05:11,244: INFO/MainProcess] celery@ubuntu ready.
[2016-11-26 22:05:31,132: INFO/MainProcess] Events of group {task} enabled by remote.
[2016-11-26 22:06:33,626: INFO/MainProcess] Received task: app.tasks.long_task[9eeed278-0e14-48ae-9762-4de447ca79ff]
[2016-11-26 22:06:35,137: INFO/MainProcess] Received task: app.tasks.long_task[8a0b46d1-abba-4bfa-844a-06ecf6f8b749]
[2016-11-26 22:07:05,259: INFO/PoolWorker-1] Task app.tasks.long_task[8a0b46d1-abba-4bfa-844a-06ecf6f8b749] succeeded in 30.118909819s: {'current': 100, 'total': 100, 'result': '完成'}
[2016-11-26 22:07:10,752: INFO/PoolWorker-3] Task app.tasks.long_task[9eeed278-0e14-48ae-9762-4de447ca79ff] succeeded in 37.123306606s: {'current': 100, 'total': 100, 'result': '完成'}
启动 flower 进程：
$ flower -A app
[I 161126 22:05:26 command:136] Visit me at http://localhost:5555
[I 161126 22:05:26 command:141] Broker: redis://localhost:6379/15
[I 161126 22:05:26 command:144] Registered tasks:
[u'app.tasks.async_send_email',
u'app.tasks.long_task',
u'celery.accumulate',
u'celery.backend_cleanup',
u'celery.chain',
u'celery.chord',
u'celery.chord_unlock',
u'celery.chunks',
u'celery.group',
u'celery.map',
u'celery.starmap']
[I 161126 22:05:26 mixins:224] Connected to redis://localhost:6379/15
[W 161126 22:05:28 control:44] 'active' inspect method failed
[W 161126 22:05:28 control:44] 'reserved' inspect method failed
[W 161126 22:05:28 control:44] 'conf' inspect method failed
网友评论
ContentDisallowed: Refusing to deserialize untrusted content of type pickle (application/x-python-serialize)
尝试了google上面所有的方案都不行,
希望博主有空可以更新下两年前的代码,谢谢
进度和更新一直是0%, PENDING,博主这是啥原因?
tasks.py 修改 ：
@Celery.task(serializer='pickle')
def async_send_email(msg):
===================分割线===========================
如果后面出现ValueError: not enough values to unpack (expected 3, got 0)这个报错
解决方法：pip install eventlet
celery -A worker -l info -P eventlet
就可以了（我用的windows会出现这个报错）