您的位置：首页 > 数据库 > Redis

flask + redis 实现Server-Sent Event demo

2017-08-28 13:52 609 查看
学习要点：

（1）web即时通讯

Web 即时通信，就是说我们可以通过一种机制在网页上立即通知用户一件事情的发生，是不需要用户刷新网页的。

常用的有Polling,Long-Polling,Server Send Event,WebSocket

Server Send Event协议：

HTTP 的头部Content-Type设置成text/event-stream就可以了。其实SSE，就是浏览器向服务器发送一个 HTTP 请求，然后服务器不断单向地向浏览器推送”信息”，这些信息的格式也非常简单，就是前缀data:加上发送的数据内容，然后以\n\n结尾。

（2）redis的订阅分发功能

Redis 发布订阅(pub/sub)是一种消息通信模式：发送者(pub)发送消息，订阅者(sub)接收消息

参考：http://www.runoob.com/redis/redis-pub-sub.html

功能：

一个点击button获取当前时间的demo



技术栈：

jQuery，flask, redis

分析：

点击button，触发事件 —-> jquery 的post方法发送url为/post，method为POST 的请求 —->/post视图函数执行，redis向time频道publish当前时间—->/stream视图函数监听time频道获得数据，转发给js中的事件源var source = new EventSource(‘/stream’);—–>最后通过innerHTML方法打印到网页

代码：

Python2/3均可

import datetime
import flask
import redis

app = flask.Flask(__name__)
app.secret_key = 'asdf'
red = redis.StrictRedis(host='localhost', port=6379, db=6)

def event_stream():
pubsub = red.pubsub()
pubsub.subscribe('time')
for message in pubsub.listen():
print (message)
yield 'data: %s\n\n' % message['data']

@app.route('/post', methods=['POST'])
def post():
now = datetime.datetime.now().replace(microsecond=0).time()
red.publish('time', u'%s' % (now.isoformat()))
return flask.Response(status=204)

@app.route('/stream')
def stream():
return flask.Response(event_stream(),
mimetype="text/event-stream")

@app.route('/')
def home():
return u"""
<!doctype html>
<title>chat</title>
<script src="http://cdn.staticfile.org/jquery/2.1.1/jquery.min.js"></script>
<style>body { max-width: 500px; margin: auto; padding: 1em; background: black; color: #fff; font: 16px/1.6 menlo, monospace; }</style>
</p><button id="time">Click for time!</button>
<pre id="out"></pre>
<script>
function sse() {
var source = new EventSource('/stream');
var out = document.getElementById('out');
source.onmessage = function(e) {
out.innerHTML = 'now-tiem:'+ e.data + '\\n' + out.innerHTML;
};
}
$('#time').click(function(){
$.post('/post');
}
);
sse();
</script>
"""

if __name__ == '__main__':
app.debug = True
app.run(threaded=True)
内容来自用户分享和网络整理，不保证内容的准确性，如有侵权内容，可联系管理员处理 点击这里给我发消息
标签： 
相关文章推荐
新的分享
章节导航