您现在的位置是:首页>文章详情

Flask + Celery 实战

泛逸舟

上一篇文章 中,我们对 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

 

评论

评论插件