基于websocket的celery任务状态监控
1. 目的
曾经想向前台实时返回Celery任务的状态监控，也查看了很多博客，但是好多也没能如愿，因此基于网上已有的博客已经自己的尝试，写了一个小的demo，实现前台实时获取后台传输的任务状态。
2. 准备
本篇文章使用的是Flask框架，安装celery，celery采用redis作为存储。同时用到了Flask-SocketIO建立websocket。同时还用到了协程库eventlet（这个是Flask-SocketIO文档建议的，链接文档）。
3. 实现
demo仿照其他例子实现了一个简单的后台任务监控。我们直接上代码吧，下面是server端代码：
# -*- utf-8 -*-
# app.py
import time
import uuid
from flask import Flask, render_template, request, make_response, jsonify
from flask_socketio import SocketIO
from celery import Celery
import eventlet
from flask_redis import FlaskRedis
eventlet.monkey_patch()
app = Flask(__name__)
app.config['BROKER_URL'] = 'redis://localhost:6379/0'
app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379/0'
app.config['CELERY_ACCEPT_CONTENT'] = ['json', 'pickle']
app.config['REDIS_URL'] = 'redis://localhost:6379/0'
socketio = SocketIO(app, async_mode='eventlet',message_queue=app.config['CELERY_RESULT_BACKEND'])
redis = FlaskRedis(app)
celery = Celery(app.name)
celery.conf.update(app.config)
#模拟后台耗时任务
@celery.task
def background_task(uid):
sid = redis.get(uid)
socketio.emit('info', {'data': 'Task starting ...', 'time': time.time() * 1000 },room=sid, namespace='/task')
socketio.sleep(4)
socketio.emit('info', {'data': 'Task running!', 'time': time.time() * 1000 }, room=sid, namespace='/task')
socketio.sleep(5)
socketio.emit('info', {'data': 'Task complete!', 'time': time.time()*1000 }, room=sid, namespace='/task')
#建立链接时把sid传到浏览器端保存。
@socketio.on('connect', namespace='/task')
def connect_host():
sid = request.sid
socketio.emit('hostadd', {'sid': sid}, room=sid, namespace='/task')
#将每一个客户端生成一个uuid存放在cookie中
@app.route('/')
def index():
if not request.cookies.get('host_uid', None):
uid = uuid.uuid1().get_hex()
response = make_response(render_template('index.html'))
response.set_cookie('host_uid', uid)
return response
return render_template('index.html')
@app.route('/task')
def start_background_task():
uid = request.cookies.get('host_uid')
background_task.delay(uid)
return 'Started'
#设置sid建立链接后浏览器将sid传送到server，并将uid与sid映射存放在redis里面，默认保留12小时
@app.route('/setsid', methods=['POST'])
def set_uid():
data = request.json
uid = request.cookies.get('host_uid')
redis.set(uid, data['sid'])
redis.expire(uid, 3600 * 12)
return jsonify({'success': True})
if __name__ == '__main__':
socketio.run(app, host='0.0.0.0', port=5000, debug=True)如果不想使用debug模式的话，可以用gunicorn运行，命令如下所示：
gunicorn --worker-class eventlet -w 1 app:app
使用上述命令需要注意，由于gunicorn负载均衡算法的限制，文档建议worker数量为1，我测试过大于1，确实会出问题。
前端代码如下，index.html:
<!DOCTYPE html>
<html>
<head>
<title>test</title>
<script type="text/javascript" src="https://cdnjs.cloudflare.com/ajax/libs/jquery/2.2.4/jquery.min.js"></script>
<script type="text/javascript" src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/2.1.0/socket.io.js"></script>
</head>
<body>
<h3>Logging</h3>
<p id="log"></p>
<button id="background">Execute</button>
<script type="text/javascript">
$(document).ready(function () {
namespace = '/task';
socket = io.connect('http://' + document.domain + ':' + location.port + namespace);
socket.on('hostadd', function(msg){
console.log(msg.sid);
$.ajax({
url: "{{ url_for('set_uid') }}",
data: JSON.stringify({ sid: msg.sid }),
type: 'post',
dataType: 'json',
contentType: "application/json; charset=utf-8"
})
});
socket.on('info', function (msg) {
console.log('Recived: ' + msg.data);
var t = new Date(msg.time);
$('#log').append('Recived: ' + t.toLocaleTimeString() + '->' + msg.data + '<br>');
});
$('#background').on('click', function(){
$.get("{{ url_for('start_background_task') }}");
});
});
</script>
</body>
</html>GitHub地址：https://github.com/junfenggoo...
相关推荐
kaixinfelix 2020-07-27
waitzkj 2020-06-20
fgleeldq 2020-06-14
xinhao 2020-06-09
也许会有hui 2020-05-03
D先生 2020-05-09
hoooooolyhu 2020-04-23
iflreey 2020-03-05
taiyanghua 2020-02-14
kanpiaoxue 2020-01-31
牧码人 2020-01-25
shawroad 2020-01-07
bluetears 2019-12-26
loviezhang 2019-12-15
Burgesszheng 2019-12-13
liusarazhang 2019-12-06