快捷導(dǎo)航

使用Python和Go實(shí)現(xiàn)服務(wù)器發(fā)送事件(SSE)

更新時(shí)間：2024年11月25日 10:33:19 作者：關(guān)山月

在當(dāng)今的交互式web應(yīng)用程序中,實(shí)時(shí)數(shù)據(jù)更新在增強(qiáng)用戶體驗(yàn)方面起著至關(guān)重要的作用,在實(shí)時(shí)通信的各種技術(shù)中,SSE在眾多解決方案脫穎而出,本文給大家介紹了使用Python和Go實(shí)現(xiàn)服務(wù)器發(fā)送事件(SSE),需要的朋友可以參考下

為什么選擇SSE？

服務(wù)器發(fā)送事件是HTML5規(guī)范的一部分，專門用于將事件從服務(wù)器推送到客戶端。它的簡單性、自動(dòng)重新連接和事件跟蹤功能使其非常適合數(shù)據(jù)流的場景。在單向數(shù)據(jù)流情況下，SSE表現(xiàn)尤其出色。

概述

SSE是一種服務(wù)器向?yàn)g覽器實(shí)時(shí)推送消息的技術(shù)。它是HTML5規(guī)范的一部分，主要涉及：

通信協(xié)議：使用HTTP。
事件對象：在瀏覽器端可用。

WebSockets也是一種實(shí)時(shí)通信技術(shù)，但它們有不同之處：

SSE	WebSockets
基于HTTP	基于TCP
單向（服務(wù)器到客戶端）	全雙工（雙向）
輕量級和簡單	更復(fù)雜的
內(nèi)置重新連接和消息跟蹤	需要手動(dòng)實(shí)現(xiàn)這些功能
文本或Base64和gzip壓縮的二進(jìn)制文件	支持各種數(shù)據(jù)類型
支持自定義事件類型	不支持自定義事件類型
HTTP/1.1或HTTP/2限制連接數(shù)量	無限連接

服務(wù)器實(shí)現(xiàn)

協(xié)議實(shí)現(xiàn)

本質(zhì)上，瀏覽器發(fā)起一個(gè)HTTP請求，服務(wù)器用HTTP狀態(tài)進(jìn)行響應(yīng)，包括以下標(biāo)頭：

Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive

SSE指定事件流的MIME類型必須為 text/event-stream ，瀏覽器不應(yīng)該緩存數(shù)據(jù)，并且連接應(yīng)該是持久的（ keep-alive ）。

消息格式

事件流是使用UTF-8編碼的文本或Base64編碼的二進(jìn)制消息，并使用gzip壓縮。每條消息由一行或多行字段組成，格式為 field-name : field-value 。每個(gè)字段以 \n 結(jié)尾。以冒號開頭的行是注釋，會被瀏覽器忽略。每個(gè)推送可以由多個(gè)消息組成，以空行分隔（ \n\n ）。

關(guān)鍵字段包括：

event ：事件類型。
id ：事件ID，瀏覽器跟蹤最后接收到的事件用來重新連接服務(wù)。
retry ：瀏覽器在連接失敗后重新嘗試連接所需的等待時(shí)間（ms）。
data ：消息數(shù)據(jù)。

示例：Python實(shí)現(xiàn)SSE

from flask import Flask, Response

app = Flask(__name__)


@app.route('/events')
def sse_handler():
    def generate():
        paragraph = [
            "Hello, this is an example of a continuous text output.",
            "It contains multiple sentences, each of which will be sent to the client as an event.",
            "This is to simulate the functionality of Server-Sent Events (SSE).",
            "We can use this method to push real-time updates.",
            "End of sample text, thank you!",
        ]

        for sentence in paragraph:
            yield f"data: {sentence}\n\n"

            import time
            time.sleep(1)

    return Response(generate(), mimetype='text/event-stream')


if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8081, debug=True)

示例：Go實(shí)現(xiàn)SSE

package main

import (
    "fmt"
    "log"
    "net/http"
    "time"
)

func main() {
    http.HandleFunc("/events", sseHandler)

    fmt.Println("Starting server on :8080")
    if err := http.ListenAndServe(":8080", nil); err != nil {
        log.Fatalf("Server error: %v", err)
    }
}

func sseHandler(w http.ResponseWriter, r *http.Request) {
    flusher, ok := w.(http.Flusher)
    if !ok {
        http.Error(w, "Streaming unsupported!", http.StatusInternalServerError)
        return
    }

    w.Header().Set("Content-Type", "text/event-stream")
    w.Header().Set("Cache-Control", "no-cache")
    w.Header().Set("Connection", "keep-alive")

    // Change the output here to a specific text
    paragraph := []string{
        "Hello, this is an example of a continuous text output.",
        "It contains multiple sentences, each of which will be sent to the client as an event.",
        "This is to simulate the functionality of Server-Sent Events (SSE).",
        "We can use this method to push real-time updates.",
        "End of sample text, thank you!",
    }

    for _, sentence := range paragraph {
        _, err := fmt.Fprintf(w, "data: %s\n\n", sentence)
        if err != nil {
            return
        }
        flusher.Flush()
        time.Sleep(1 * time.Second) // Wait 1 second before sending the next piece of text
    }
}

瀏覽器API?

在客戶端，JavaScript的 EventSource API允許您創(chuàng)建一個(gè) EventSource 對象來偵聽服務(wù)器發(fā)送的事件。一旦連接上，服務(wù)器就可以向?yàn)g覽器發(fā)送事件消息。瀏覽器通過監(jiān)聽 onmessage 、 onopen 和 onerror 事件來處理這些消息。

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>SSE Example ??</title>
</head>
<body>
    <h1>Server-Sent Events Example ??</h1>
    <div id="messages"></div>
    <script>
        window.onload = function() {
            if (typeof(EventSource) !== "undefined") {
                const eventSource = new EventSource('/events');

                eventSource.onmessage = function(event) {
                    const newElement = document.createElement("p");
                    newElement.textContent = "Message: " + event.data;

                    document.getElementById("messages").appendChild(newElement);
                };

                eventSource.onerror = function(event) {
                    console.error("Error occurred: ", event);
                    const newElement = document.createElement("p");
                    newElement.textContent = "An error occurred while connecting to the event source.";
                    document.getElementById("messages").appendChild(newElement);
                    eventSource.close(); 
                };
            } else {
                document.getElementById("messages").textContent = "Sorry, your browser does not support server-sent events...";
            }
        };
    </script>
</body>
</html>

SSE調(diào)試

目前，許多流行的工具，如Postman、Insomnia、Bruno和ThunderClient缺乏對服務(wù)器發(fā)送事件SSE的足夠支持。在開發(fā)過程中，這種限制會讓人非常沮喪。幸運(yùn)的是，我最近遇到了EchoAPI，這個(gè)工具提供了出色的SSE調(diào)試功能。這個(gè)發(fā)現(xiàn)極大地改善了我的工作流程，提高了效率和生產(chǎn)力。