代碼段

App.py 主文件

import datetime
from flask import Flask, session, redirect, Response, request, render_template, url_for, flash
from redis import StrictRedis

app = Flask(__name__)

# session cookie密鑰
app.secret_key = 'pardon110'

# 連接redis數據庫,默認是零號庫,可隨便更改
rds = StrictRedis(db=3)

# 消息生成器
def event_stream():
 # 從數據庫連接上獲取發布訂閱管理對象實例
 pub = rds.pubsub()
 # 在管理訂閱(建立通道)頻道
 pub.subscribe('chat')
 # 監聽頻道信息
 for message in pub.listen():
 print(type(message['data']), type(message), message)
 # 只響應有消息的（字節），首次無消息返回的為int狀態碼對象，直接忽略
 if isinstance(message['data'], bytes):
 # 轉為utf8字符串，發送 SSE（Server Send Event）協議格式的數據
 yield 'data: %snn' % message['data'].decode()

# 首次訪問需要登錄
@app.route('/login', methods=['GET', 'POST'])
def login():
 if request.method == 'POST':
 session['username'] = request.form['user']
 # 重定向到home處理器
 flash('您已經成功登錄！')
 return redirect(url_for('home'))
 return '<p><strong>登錄</strong></p><form action="" method="post">用戶名: <input name="user">'

# 接收js發送過來的消息
@app.route('/post', methods=['POST'])
def post():
 # 獲取表單提交內容
 message = request.form['message']
 # 獲取當前請求對象的session實例
 user = session.get('username', 'anonymous')
 # 返回一個指定字段的時間值
 now = datetime.datetime.now().replace(microsecond=0).time()
 # 通過頻道發布消息
 rds.publish('chat', u'[%s] %s: %s' % (now.isoformat(), user, message))
 # 響應對象
 return Response(status=204)

# 事件流接口
@app.route('/stream')
def stream():
 return Response(event_stream(), mimetype="text/event-stream")

@app.route('/')
@app.route('/<name>')
def home(name=None):
 # 通過路由參數或querystring動注冊為當前用戶
 if name or len(request.args) > 0:
 session['username'] = name if name else request.args.get(
 'user', '')
 # 消息閃現（存儲在session內，模板頁用完即丟）
 flash(session['username']+'已經成功登錄，加入聊天室！')
 # 否則強制用戶登錄
 if 'username' not in session:
 return redirect('/login')

 # 模板渲染
 data = {
 "username": session['username'],
 "tip": "正在聊天中..."
 }
 # 關鍵字參數解包，返回元組（框架會自動解析為一個完整的response對象）
 return render_template('home.html', **data), 200

@app.route('/logout')
def logout():
 # 清空當前session信息
 session.pop('username', None)
 return redirect(url_for('login'))

if __name__ == '__main__':
 # app.debug = True
 # falsk1.0.2默認多線程開啟，默認端口是5000
 app.run(host='0.0.0.0', debug=True)

sse.js 文件

// vue實例
var vm = new Vue({
 el: '#out',
 // 自定義定界符，避免vue定界符與jinja2沖突，改變vue起始定界
 delimiters: ['@{', '}}'],
 // 實例屬性
 data: {
 items: []
 },
 mounted() {
 // 監聽服務端消息
 sse()
 }
})

// 通過EventSource對象與后端建立連接
function sse() {
 var source = new EventSource('/stream')
 // 監聽消息事件（與nodejs用data事件sock流
 source.onmessage = function (e) {
 // XSS in chat is fun
 // 動態渲染頁面（代理vue實例數據操作）
 vm.$data.items.push(e.data)
 }
}

// ajax與后端交互，向服務端發送消息
$('#in').keyup(function (e) {
 // 回車keycode為13
 if (e.keyCode == 13) {
 $.post('/post', { 'message': $(this).val() })
 $(this).val('')
 }
})

// 自動消失閃現
$(document).ready(function () {
 setTimeout(() => {
 $(".flashes").fadeOut(3000);
 })
})

home.html 模板頁面

<html lang="en">

<head>
 <meta charset="UTF-8">
 <meta name="viewport" content="width=device-width, initial-scale=1.0">
 <meta http-equiv="X-UA-Compatible" content="ie=edge">
 <title>基于redis簡單聊天室</title>
 <style>
 body {
 max-width: 500px;
 margin: auto;
 padding: 1em;
 background: black;
 color: #fff;
 font: 16px/1.6 menlo, monospace;
 }
 </style>
</head>
<body>
 <!-- jinja2模板渲染 -->
 <p><b>Hi! {{ username }} {{ tip }}</b></p>
 <hr>
 <!-- jinja2 渲染閃現消息 -->
 {% with messages = get_flashed_messages() %}
 {% if messages %}
 <ul class=flashes>
 {% for message in messages %}
 <li>{{ message }}</li>
 {% endfor %}
 </ul>
 {% endif %}
 {% endwith %}

 <!-- 輸出聊天內容 -->
 <div id="out">
 <!-- vue組件渲染 -->
 <span v-for="(item,index) in items" :key="index">@{ item }}<br></span>
 </div>
 <!-- 輸入消息框 -->
 <div>
 <p>Message: <input id="in" placeholder="回車發送信息" /></p>
 </div>

 <!-- cdn資源 -->
 <script src="http://cdn.staticfile.org/jquery/2.1.1/jquery.min.js"></script>
 <script src="https://cdn.jsdelivr.net/npm/vue@2.6.10/dist/vue.js"></script>

 <!-- vue及sse相關代碼 -->
 <script src="/static/sse.js"></script>

</body>

</html>