快捷導(dǎo)航

Flask框架里面sse的使用示例

更新時間：2025年05月20日 09:24:22 作者：zwjapple

本文主要介紹了Flask框架里面sse的使用示例,文中通過示例代碼介紹的非常詳細,對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧

可以在一個 Flask 應(yīng)用中注冊多個 Blueprint，每個 Blueprint 可以對應(yīng)一個 SSE 接口。例如：

from flask import Flask, Response, Blueprint
import time

app = Flask(__name__)

bp1 = Blueprint('sse1', __name__)
bp2 = Blueprint('sse2', __name__)

@bp1.route('/sse1')
def sse1():
    def generate():
        while True:
            yield 'data: {}\n\n'.format(time.time())
            time.sleep(1)
    return Response(generate(), mimetype='text/event-stream')

@bp2.route('/sse2')
def sse2():
    def generate():
        while True:
            yield 'data: {}\n\n'.format(time.strftime('%Y-%m-%d %H:%M:%S'))
            time.sleep(1)
    return Response(generate(), mimetype='text/event-stream')

app.register_blueprint(bp1)
app.register_blueprint(bp2)

if __name__ == '__main__':
    app.run(debug=True)

在這個例子中，我們創(chuàng)建了兩個 Blueprint，分別對應(yīng) `/sse1` 和 `/sse2` 接口。每個 Blueprint 中的函數(shù)都返回一個 SSE 數(shù)據(jù)流，使用 `Response` 類型的響應(yīng)對象封裝。最后在 Flask 應(yīng)用中注冊這兩個 Blueprint，即可同時開啟兩個 SSE 接口。

from flask import Flask, Response, request

app = Flask(__name__)

# 存儲所有的訂閱者
subscribers = {}

# 訂閱接口，每個瀏覽器對應(yīng)一個topic
@app.route('/subscribe/<topic>')
def subscribe(topic):
    def stream():
        # 將當前請求的客戶端添加到訂閱者列表中
        subscribers.setdefault(topic, []).append(stream)
        # 無限循環(huán)，等待新的消息
        while True:
            # 等待新的消息
            message = request.args.get('message')
            if message:
                # 向所有訂閱者推送新的消息
                for subscriber in subscribers.get(topic, []):
                    subscriber.put(message)
            # 休眠一段時間，減少服務(wù)器壓力
            time.sleep(1)

    # 設(shè)置響應(yīng)頭，告訴客戶端這是一個SSE流
    response = Response(stream(), mimetype='text/event-stream')
    response.headers['Cache-Control'] = 'no-cache'
    response.headers['Connection'] = 'keep-alive'
    return response

# 推送接口，向指定topic的所有訂閱者推送消息
@app.route('/publish/<topic>')
def publish(topic):
    # 獲取要推送的消息
    message = request.args.get('message')
    if message:
        # 向所有訂閱者推送新的消息
        for subscriber in subscribers.get(topic, []):
            subscriber.put(message)
    return 'OK'

在上面的代碼中，我們定義了兩個接口，`/subscribe/<topic>`用于訂閱指定的topic，`/publish/<topic>`用于向指定topic的所有訂閱者推送消息。
在訂閱接口中，我們將當前請求的客戶端添加到訂閱者列表中，并通過一個無限循環(huán)等待新的消息。當有新的消息到達時，我們會向所有訂閱者推送這條消息。
在推送接口中，我們獲取要推送的消息，并向指定topic的所有訂閱者推送這條消息。
使用這個代碼，你可以輕松地實現(xiàn)一個簡單的SSE推送服務(wù)，每個瀏覽器對應(yīng)一個topic。

import uuid
from flask import Flask, jsonify, request, Response, g
from flask_cors import CORS
from flask_sse import sse
import time
import json

app = Flask(__name__)
cros = CORS(app)
app.config['REDIS_URL'] = 'redis://localhost'
app.register_blueprint(sse, url_prefix='/stream')
# SSE 推送函數(shù)


# SSE 推送路由


@app.route('/register', methods=["GET"])
def register():
    # 獲取客戶端標識符
    client_id = str(uuid.uuid4())

    # 返回 SSE 響應(yīng)
    return jsonify({"client_id": client_id})

# SSE 推送路由


@app.route('/sse', methods=['POST'])
def stream():
    # 獲取客戶端標識符
    data = request.get_json()
    client_id = data['clientId']
    print("client_id", client_id)

    def aa():
        # 循環(huán)發(fā)送 SSE 數(shù)據(jù)
        for i in range(10):
            data = 'Hello, %s!' % client_id+str(i)
            sse.publish(data, channel=client_id, type='message')
            time.sleep(1)
        sse.publish("end", channel=client_id, type='message')
     # 返回 SSE 響應(yīng)
    response = Response(aa(), mimetype='text/event-stream')
    response.headers.add('Cache-Control', 'no-cache')
    response.headers.add('Connection', 'keep-alive')
    response.headers.add('X-Accel-Buffering', 'no')
    return response


if __name__ == '__main__':
    app.run(debug=True, port=5000)

在上面的代碼中，我們使用了 Flask-SSE 擴展來管理 SSE 通道。我們在 `/register` 路由中為每個客戶端創(chuàng)建了唯一的標識符，并將其存儲在請求環(huán)境中。在 `/sse` 路由中，我們使用 `sse.publish` 方法發(fā)送 SSE 數(shù)據(jù)。

到此這篇關(guān)于Flask框架里面sse的使用示例的文章就介紹到這了,更多相關(guān)Flask使用sse內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: