FlaskRedis建構實時訊息系統與ETag應用

– 11 分鐘閱讀

在現今的網路應用中，實時訊息傳輸和高效的資源管理至關重要。本文將深入探討如何利用 Flask 和 Redis 建構一個簡易的實時訊息系統，並講解 ETag 在 HTTP 請求中的應用，包含條件式請求、樂觀並發控制，以及非同步任務處理。藉由結合 Redis 的釋出/訂閱功能和 Flask 的路由機制，我們可以實作一個允許使用者向指定頻道傳送訊息，並透過另一個端點實時接收訊息的系統。同時，我們將探討如何利用 ETag 提升效能、減少網路負擔，並確保資料一致性，包含條件式請求的實作、樂觀並發控制的策略，以及非同步任務處理的技巧。

Flask 是一個輕量級的 Web 框架，而 Redis 則是一個高效能的鍵值儲存系統，兩者結合可以輕鬆地搭建實時訊息系統。我們可以利用 Flask 建立兩個端點：一個用於傳送訊息，另一個用於接收訊息。傳送訊息的端點會將訊息釋出到 Redis 的指定頻道，而接收訊息的端點則會訂閱該頻道，並將接收到的訊息透過事件流（Event Stream）傳回給客戶端。此外，ETag 作為 HTTP 標準的一部分，允許客戶端使用其快取進行條件式請求，從而限制頻寬和伺服器資源的使用。伺服器在回應中包含 ETag 標頭，客戶端可以使用 If-None-Match 標頭來決定是否下載頁面，如果 ETag 匹配，則傳回 304 Not Modified，客戶端可以使用快取副本。更進一步，ETag 可用於實作樂觀並發控制，確保在多個使用者同時存取相同資源時，資料的一致性。最後，我們將探討如何結合 Flask 和 UUID 處理非同步任務，以提升應用程式的效能和反應速度。

概述

本文將介紹如何使用 Flask 和 Redis 建立一個簡單的實時訊息系統。這個系統允許使用者向指定的頻道傳送訊息，並且可以透過另一個端點實時接收這些訊息。

端點一：接收訊息

首先，我們定義了一個端點 /message/<channel>，它使用 HTTP POST 方法接收 JSON 格式的訊息。這個端點需要包含 source 和 content 兩個欄位，分別代表訊息來源和內容。

@application.route("/message/<channel>", methods=['POST'])
def send_message(channel):
    data = flask.request.json
    if not data or 'source' not in data or 'content' not in data:
        flask.abort(400)
    
    r = redis.Redis()
    r.publish(channel, "<{}> {}".format(data["source"], data["content"]))
    return "", 202

端點二：傳送訊息

另一個端點也是 /message/<channel>，但它使用 HTTP GET 方法，傳回一個事件流（Event Stream），用於實時接收訊息。這個端點傳回的內容型別是 text/event-stream，它是一種特殊的 MIME 型別，允許伺服器向客戶端推播事件。

def get_messages(channel):
    return flask.Response(
        flask.stream_with_context(stream_messages(channel)),
        mimetype='text/event-stream'
    )

事件流生成器

stream_messages 函式是一個生成器，它不斷地從 Redis 中接收訊息，並將其 yield 出去，以便 Flask 可以將其作為事件流傳回給客戶端。

def stream_messages(channel):
    # 從 Redis 中接收訊息並 yield 出去
    #...

執行伺服器

要執行這個應用程式，需要使用一個支援多個連線的 Web 伺服器，如 uWSGI。可以使用以下命令執行伺服器：

$ uwsgi --http :5000 --master --workers 10 --wsgi-file http.py

測試

在另一個終端中，可以使用 HTTP 客戶端連線到事件流，並在另一個終端中傳送訊息。

# 連線到事件流
HTTP/1.1 200 OK
Content-Type: text/event-stream; charset=utf-8

# 傳送訊息
HTTP/1.1 202 ACCEPTED
Content-Length: 0
Content-Type: text/html; charset=utf-8

# 接收到訊息
data: <jd> it works

使用 ETag 進行條件請求

ETag（實體標籤）是 HTTP 標準的一部分，允許客戶端使用其快取進行條件請求，從而限制頻寬和伺服器資源的使用。當客戶端傳送請求給伺服器時，伺服器可以在回應中包含 ETag 標頭。ETag 的生成方法包括使用資源內容的碰撞抗性雜湊函式、最後修改時間戳記的雜湊或甚至只是修訂號。

ETag 標頭示例

HTTP/1.1 200 OK
Accept-Ranges: bytes
Connection: Keep-Alive
Content-Length: 15919
Content-Type: text/html; charset=UTF-8
Date: Tue, 12 Sep 2017 09:17:26 GMT
ETag: "3e2f-558f9d5587b40"
Keep-Alive: timeout=5, max=100
Last-Modified: Tue, 12 Sep 2017 08:28:53 GMT
Server: Apache

使用 If-None-Match 標頭

客戶端可以使用 If-None-Match 標頭來決定是否下載頁面。示例：

HTTP/1.1 304 Not Modified
Connection: Keep-Alive
Date: Tue, 12 Sep 2017 09:19:28 GMT
ETag: "3e2f-558f9d5587b40"
Keep-Alive: timeout=5, max=100
Server: Apache

如果 URL 的 ETag 值與 If-None-Match 標頭中的值匹配，則 HTTP 狀態碼傳回為 304 Not Modified，且不傳回任何內容。客戶端因此知道內容未在伺服器上改變，可以使用其快取的副本。

計算 ETag

理想情況下，計算 ETag 應該盡可能簡單，以節省 CPU 使用率和網路頻寬。對於檔案，簡單的 ETag 可以從檔案最後修改時間戳記和大小計算，因為這些資訊可以從作業系統中以低成本獲得。更強大的 ETag 可以透過計算即將傳回的資料的 MD5 雜湊來生成。但是，這可能更昂貴，尤其是當資料量大時。

Flask 應用示例

以下示例展示瞭如何使用 ETag 標頭來避免傳回請求的內容：

import flask
from werkzeug.exceptions import HTTPException

app = flask.Flask(__name__)

class NotModified(HTTPException):
    code = 304

@app.route("/", methods=['GET'])
def get_index():
    # 因為內容始終相同，因此只需一個 ETag
    etag = "hword"
    #...

內容解密：

在這個示例中，我們定義了一個 NotModified 類別，繼承自 HTTPException，並設定其程式碼為 304。然後，在 get_index 函式中，我們計算 ETag 值，並使用它來決定是否傳回內容。如果 ETag 值與客戶端提供的值匹配，則傳回 304 Not Modified 狀態碼，否則傳回內容。

圖表翻譯：

  sequenceDiagram
    participant 客戶端
    participant 伺服器
    客戶端->>伺服器: GET 請求
    伺服器->>客戶端: 200 OK，包含 ETag 標頭
    客戶端->>伺服器: GET 請求，包含 If-None-Match 標頭
    alt ETag 值匹配
        伺服器->>客戶端: 304 Not Modified
    else ETag 值不匹配
        伺服器->>客戶端: 200 OK，包含更新的內容
    end

這個圖表展示了客戶端和伺服器之間的互動過程，包括 ETag 的計算和比較。

HTTP 請求頭部：If-Match 和 If-None-Match

在 HTTP 協定中，If-Match 和 If-None-Match 是兩個重要的請求頭部，分別用於條件式請求，以確保客戶端和伺服器之間的資料一致性。

If-Match

If-Match 頭部用於指定一個或多個實體標籤（ETag），只有當伺服器上的資源的 ETag 匹配其中一個指定的 ETag 時，伺服器才會處理這個請求。這通常用於確保客戶端擁有的資源版本是最新的，如果版本不匹配，伺服器會傳回 412 預期失敗（Precondition Failed）狀態碼。

If-None-Match

If-None-Match 頭部與 If-Match 相反，它指定一個或多個實體標籤，如果伺服器上的資源的 ETag 不匹配任何一個指定的 ETag，伺服器才會處理這個請求。如果 ETag 匹配，伺服器會傳回 304 未修改（Not Modified）狀態碼，指示客戶端可以使用其本地快取的版本。

Flask 實作

以下是使用 Flask 框架實作 If-Match 和 If-None-Match 的簡單範例：

from flask import Flask, request, Response

app = Flask(__name__)

# 定義一個路由
@app.route('/', methods=['GET'])
def index():
    # 定義一個 ETag 值
    ETAG = "hword"
    
    # 取得 If-Match 和 If-None-Match 頭部
    if_match = request.headers.get("If-Match")
    if_none_match = request.headers.get("If-None-Match")
    
    # 處理 If-Match
    if if_match is not None and if_match!= ETAG:
        return Response(status=412)  # 412 預期失敗
    
    # 處理 If-None-Match
    if if_none_match is not None and if_none_match == ETAG:
        return Response(status=304)  # 304 未修改
    
    # 回應內容和 ETag
    return Response("hello world", headers={"ETag": ETAG})

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

單元測試

為了確保這個實作是正確的，我們可以寫一些單元測試。以下是使用 unittest 框架的範例：

import unittest
from yourapplication import application  # 匯入你的 Flask 應用

class TestApp(unittest.TestCase):
    def test_get_index(self):
        test_app = application.test_client()
        result = test_app.get('/')
        self.assertEqual(200, result.status_code)
        
    def test_get_index_if_match_positive(self):
        test_app = application.test_client()
        result = test_app.get('/', headers={"If-Match": "hword"})
        self.assertEqual(200, result.status_code)
        
    def test_get_index_if_match_negative(self):
        test_app = application.test_client()
        result = test_app.get('/', headers={"If-Match": "foobar"})
        self.assertEqual(412, result.status_code)
        
    def test_get_index_if_none_match_positive(self):
        test_app = application.test_client()
        result = test_app.get('/', headers={"If-None-Match": "hword"})
        self.assertEqual(304, result.status_code)
        
    def test_get_index_if_none_match_negative(self):
        test_app = application.test_client()
        result = test_app.get('/', headers={"If-None-Match": "foobar"})
        self.assertEqual(200, result.status_code)

if __name__ == '__main__':
    unittest.main()

這些測試涵蓋了不同情況下的 If-Match 和 If-None-Match 行為，確保應用按照預期工作。

使用 ETag 實作樂觀並發控制

ETag（實體標籤）是一種 HTTP 標頭，用於標識資源的版本。當使用者端向伺服器請求資源時，伺服器會傳回資源的 ETag 值。使用者端可以在後續請求中包含此 ETag 值，以便伺服器判斷資源是否已經修改。

Flask 中使用 ETag

以下是一個使用 Flask 框架實作 ETag 的例子：

from flask import Flask, request, jsonify
import random

app = Flask(__name__)

# 初始化變數
VALUE = "hello"
ETAG = random.randint(1000, 5000)

# 定義一個自定義的 NotModified 例外類別
class NotModified(Exception):
    code = 304

# 定義 GET 端點
@app.route("/", methods=["GET"])
def get_value():
    return jsonify({"value": VALUE, "ETag": ETAG})

# 定義 POST 端點
@app.route("/", methods=["POST"])
def update_value():
    global VALUE, ETAG
    # 更新 VALUE 和 ETag
    VALUE = "new_value"
    ETAG = random.randint(1000, 5000)
    return jsonify({"value": VALUE, "ETag": ETAG})

# 定義 PUT 端點
@app.route("/", methods=["PUT"])
def put_value():
    global VALUE, ETAG
    # 檢查 ETag 是否匹配
    if request.headers.get("If-Match")!= ETAG:
        raise NotModified
    # 更新 VALUE 和 ETag
    VALUE = "new_value"
    ETAG = random.randint(1000, 5000)
    return jsonify({"value": VALUE, "ETag": ETAG})

if __name__ == "__main__":
    app.run()

在這個例子中，我們定義了三個端點：GET、POST 和 PUT。GET 端點傳回資源的值和 ETag。POST 端點更新資源的值和 ETag。PUT 端點檢查 ETag 是否匹配，如果匹配則更新資源的值和 ETag。

測試 ETag

我們可以使用 unittest 框架測試 ETag 的實作：

import unittest
from your_app import app

class TestETag(unittest.TestCase):
    def test_get_etag(self):
        response = app.test_client().get("/")
        self.assertEqual(response.status_code, 200)
        self.assertIn("ETag", response.headers)

    def test_post_etag(self):
        response = app.test_client().post("/")
        self.assertEqual(response.status_code, 200)
        self.assertIn("ETag", response.headers)

    def test_put_etag(self):
        response = app.test_client().put("/", headers={"If-Match": "12345"})
        self.assertEqual(response.status_code, 304)

if __name__ == "__main__":
    unittest.main()

在這個例子中，我們定義了三個測試案例：測試 GET 端點傳回的 ETag、測試 POST 端點更新的 ETag 和測試 PUT 端點檢查 ETag 的實作。

HTTP 請求中的 ETag 機制

HTTP 協定中，ETag（Entity Tag）是一種用於判斷資源版本是否變化的機制。它允許客戶端和伺服器之間進行資源版本的對比，以確保在進行修改或更新時，不會因為版本衝突而導致錯誤。

ETag 的工作原理

  1. 伺服器生成 ETag：當客戶端請求一個資源時，伺服器會根據該資源的內容或其他相關因素生成一個唯一的 ETag 值，並將其包含在 HTTP 回應頭中。
  2. 客戶端儲存 ETag：客戶端收到回應後，會儲存這個 ETag 值，以便下次請求時使用。
  3. If-Match 和 If-None-Match：客戶端在傳送請求時，可以在請求頭中包含 If-Match 或 If-None-Match 欄位，分別用於條件式請求：
    • If-Match: 只有當伺服器上的 ETag 值與客戶端提供的值匹配時，伺服器才會處理這個請求。
    • If-None-Match: 只有當伺服器上的 ETag 值與客戶端提供的值不匹配時，伺服器才會處理這個請求。

實作 ETag 機制的 Flask 應用

以下是一個簡單的 Flask 應用，展示瞭如何實作 ETag 機制：

from flask import Flask, request, Response
import random
import unittest

app = Flask(__name__)

# 全域性變數儲存 ETag 和資源值
ETAG = "initial_etag"
VALUE = b"initial_value"

def check_etag(exception_class):
    if_match = request.headers.get("If-Match")
    if if_match is not None and if_match!= ETAG:
        raise exception_class
    
    if_none_match = request.headers.get("If-None-Match")
    if if_none_match is not None and if_none_match == ETAG:
        raise exception_class

@app.route("/", methods=['GET'])
def get_index():
    check_etag(NotModified)
    return Response(VALUE, headers={"ETag": ETAG})

@app.route("/", methods=['PUT'])
def put_index():
    global ETAG, VALUE
    check_etag(exceptions.PreconditionFailed)
    
    ETAG += str(random.randint(3, 9))
    VALUE = request.data
    return Response(VALUE, headers={"ETag": ETAG})

class TestApp(unittest.TestCase):
    def test_put_index_if_match_positive(self):
        test_app = app.test_client()
        resp = test_app.get()
        etag = resp.headers["ETag"]
        new_value = b"foobar"
        result = test_app.put(headers={"If-Match": etag}, data=new_value)
        # 進行斷言以驗證結果

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

測試 ETag 機制

上述程式碼中，我們定義了一個簡單的 Flask 應用，它支援 GET 和 PUT 請求。GET 請求傳回目前的資源值和 ETag，而 PUT 請求更新資源值並生成新的 ETag。

測試類別 TestApp 中的 test_put_index_if_match_positive 方法展示瞭如何使用 test_client 來模擬客戶端請求，並驗證 ETag 機制的正確性。

優先考慮的HTTP請求方法：PUT與ETag的應用

在Web開發中，尤其是在RESTful API設計中，PUT請求方法被用於更新資源。然而，在多個使用者或客戶端同時存取相同資源時，可能會發生衝突。為了避免這種情況，可以使用ETag（實體標籤）和If-Match標頭來實作樂觀的並發控制。

ETag的作用

ETag是一個識別資源版本的字串，通常由伺服器根據資源的內容生成。當客戶端傳送GET請求時，伺服器會在回應中包含ETag標頭，讓客戶端知道目前資源的版本。

If-Match標頭

當客戶端要更新資源時，可以在PUT請求中包含If-Match標頭，並將之前從伺服器收到的ETag值作為其值。如果伺服器上的資源版本與客戶端提供的ETag值不匹配，伺服器會傳回412 Precondition Failed狀態碼，指示客戶端重新取得最新版本的資源並重試更新操作。

實作重試機制

為了實作這種樂觀的並發控制模型，可以在客戶端實作一個重試機制。以下是一個簡單的例子：

while True:
    resp = client.get()
    etag = resp.headers["ETag"]
    new_data = do_something(resp.data)
    resp = client.put(data=new_data, headers={"If-Match": etag})
    
    if resp.status_code == 200:
        break
    elif resp.status_code == 412:
        # 重新取得最新版本的資源並重試更新操作
        continue

在這個例子中，客戶端會不斷嘗試更新資源，直到更新操作成功（200狀態碼）或遇到其他錯誤。

單元測試

為了確保這種機制的正確性，可以撰寫單元測試。以下是一個簡單的例子：

def test_put_index_if_match_positive(self):
    test_app = app.test_client()
    result = test_app.put(headers={"If-Match": "correct_etag"})
    self.assertEqual(200, result.status_code)

def test_put_index_if_match_negative(self):
    test_app = app.test_client()
    result = test_app.put(headers={"If-Match": "wrong_etag"})
    self.assertEqual(412, result.status_code)

在這個例子中，測試函式會檢查當If-Match標頭中的ETag值正確或不正確時，伺服器傳回的狀態碼是否正確。

9.4 非同步HTTP API

在撰寫HTTP API時，傳回200 OK狀態碼以表示請求成功是一種常見的做法。然而，這種方法可能會導致問題，因為如果觸發的動作需要較長的時間來執行，則會阻塞呼叫者，增加失敗的風險。特別是當連線持續太久（例如幾秒鐘）且網路發生問題時，連線可能會被中斷。在這種情況下，呼叫者必須重試請求。如果這種問題發生在成千上萬的客戶端，則意味著大量的CPU時間和網路頻寬被浪費。

為了避免這些問題，可以透過使長時間執行的操作變為非同步來實作。這可以透過傳回202 Accepted狀態碼來實作，該狀態碼表示請求已被接受並正在被處理。然後，可以使用另一個非同步程式來處理請求。

以下是一個使用Flask實作非同步工作處理的應用示例。該API允許任何客戶端使用其求和服務：傳遞數字給該服務，它將對數字進行求和。客戶端可以稍後要求結果，一旦它準備好。

import queue
import threading
import uuid
import flask
from werkzeug import routing

app = flask.Flask(__name__)

JOBS = queue.Queue()
RESULTS = {}

class UUIDConverter(routing.BaseConverter):
    @staticmethod
    def to_python(value):
        try:
            return uuid.UUID(value)
        except ValueError:
            raise routing.ValidationError

#...

內容解密：

在上述程式碼中，我們定義了一個Flask應用，並使用queue模組實作了一個工作佇列。當客戶端傳送請求時，我們將其新增到佇列中，並傳回202 Accepted狀態碼以表示請求已被接受。然後，另一個非同步程式可以從佇列中取出工作並進行處理。

圖表翻譯：

  flowchart TD
    A[客戶端傳送請求] --> B[新增到工作佇列]
    B --> C[傳回202 Accepted狀態碼]
    C --> D[非同步程式處理工作]
    D --> E[傳回結果給客戶端]

圖表說明：

上述流程圖顯示了客戶端傳送請求、新增到工作佇列、傳回202 Accepted狀態碼、非同步程式處理工作以及傳回結果給客戶端的整個過程。這種非同步處理方式可以有效地避免由於長時間執行的操作而導致的阻塞和失敗問題。

使用 Flask 和 UUID 處理非同步任務

在這個範例中，我們將使用 Flask 框架和 UUID 來處理非同步任務。首先，我們需要安裝必要的套件，包括 flask 和 uuid。

從商業價值與使用者經驗視角來看，善用 ETag 和非同步處理機制能顯著提升網站效能和使用者經驗。分析 Flask 與 Redis 建構實時訊息系統、ETag 條件請求與樂觀鎖，以及非同步任務處理的實作細節，可以發現這些技術有效降低伺服器負載、減少網路延遲，並提升系統的擴充套件性與穩定性。然而，ETag 的計算成本以及非同步任務的複雜度管理仍是需要權衡的技術限制。展望未來，隨著 HTTP 標準的演進和非同步程式設計模型的普及，預期 ETag 和非同步 API 將在建構高互動、低延遲的 Web 應用中扮演更關鍵的角色。對於追求高效能和優異使用者經驗的開發者而言，深入理解並應用這些技術至關重要。玄貓認為，掌握這些核心技術，能讓開發者在競爭激烈的市場中脫穎而出，創造更大的商業價值。

玄貓

玄貓

技術愛好者，專注於分享程式開發、雲端技術與 AI 應用的心得體會。