FastAPI-WebSockets mit mehr als einem Worker: jede Verbindung erreichen

Mit FastAPI sehen WebSockets nach ein paar Zeilen aus. Sie fügen mit @app.websocket einen Endpoint hinzu, halten die offenen Verbindungen in einer Liste und senden an jede davon, wenn etwas passiert. Auf Ihrem Laptop funktioniert das auf Anhieb: Die Bestellung wechselt auf „versendet“, und jede geöffnete Seite sieht es.

Dann deployen Sie mit mehr als einem Worker, und manche Seiten aktualisieren sich nicht mehr. Nicht alle, und genau das macht den Fehler so schwer zu finden. Dieser Beitrag erklärt, warum, was nötig ist, damit WebSockets über mehrere Worker und Server hinweg funktionieren, und wie Sie Live-Updates aus FastAPI senden, ohne überhaupt einen WebSocket-Server zu betreiben.

Warum es mit einem Worker funktioniert und mit vier nicht

Die übliche erste Version hält die Verbindungen im Speicher:

connections: list[WebSocket] = []

@app.websocket("/ws")
async def updates(websocket: WebSocket):
    await websocket.accept()
    connections.append(websocket)
    try:
        while True:
            await websocket.receive_text()
    except WebSocketDisconnect:
        connections.remove(websocket)

async def broadcast(message: str):
    for connection in connections:
        await connection.send_text(message)

Diese Liste lebt in einem einzigen Prozess. In Produktion laufen mehrere, als Worker eines Servers oder als Kopien Ihrer App hinter einem Load Balancer, und jeder hat seine eigene Liste. Ein Browser verbindet sich mit dem Worker, den der Load Balancer gerade ausgewählt hat. Wenn eine Anfrage auf Worker 2 broadcast() aufruft, bekommen es nur die Browser, die mit Worker 2 verbunden sind. Die anderen erfahren nie davon.

Mit einem Worker ist der Fehler unsichtbar. Mit vier Workern verpassen ungefähr drei von vier Seiten jedes Update, und welche es sind, ändert sich jedes Mal, wenn sich jemand neu verbindet.

Was es über mehrere Worker hinweg braucht

Um jede Verbindung zu erreichen, müssen die Worker jedes Event miteinander teilen:

• Ein gemeinsamer Kanal zwischen den Workern. Jeder Worker abonniert einen Pub/Sub-Kanal, meist Redis, und jeder Broadcast wird dort veröffentlicht. Jeder Worker sendet ihn dann an die Browser, die er hält. Das ist ein weiterer Dienst, den Sie betreiben müssen, und jedes Event macht jetzt einen zusätzlichen Hop.
• Sticky Sessions oder nicht. Ein WebSocket bleibt bei dem Worker, auf dem er geöffnet wurde, die HTTP-Anfragen drumherum aber nicht. Alles, was ein Worker über eine Verbindung festhält, etwa wer der Nutzer ist oder welchen Räumen er beigetreten ist, muss funktionieren, ganz gleich, welcher Worker als Nächstes antwortet.
• Wissen, wer sich verbindet. Ein Browser kann bei einem WebSocket keine Header setzen, also muss der Endpoint das Session-Cookie oder ein Token in der Adresse lesen, prüfen und entscheiden, in welchen Räumen dieser Nutzer zuhören darf.
• Deploys und Neustarts. Ein Neustart eines Workers trennt jede Verbindung, die er hält, und alle verbinden sich gleichzeitig neu. Eine Seite, die weg war, verpasst, was in der Zwischenzeit gesendet wurde, es sei denn, etwas hebt die letzten Events auf, damit sie aufholen kann.
• Timeouts im Weg. Load Balancer und Proxys schließen Verbindungen, die inaktiv wirken, also braucht der Endpoint einen Heartbeat, und der Proxy muss überhaupt erst so eingerichtet sein, dass er WebSockets durchlässt.

Nichts davon ist exotisch. Es ist eine Echtzeit-Schicht, die Sie neben Ihrer App aufbauen und am Laufen halten müssen, und jeder Dienst, der Live-Updates haben soll, baut sie von Neuem.

Oder eine Echtzeit-Cloud hält die Verbindungen

Ein gehosteter Echtzeit-Dienst nimmt die Verbindungen aus Ihren Workern heraus. Die Browser verbinden sich mit dem Dienst. Ihre FastAPI-Endpoints senden jedes Event in einer kurzen HTTP-Anfrage, ganz gleich, welcher Worker die Anfrage bearbeitet, und der Dienst stellt es jeder Seite zu, die es bekommen soll. Fügen Sie Worker hinzu, starten Sie sie neu oder verschieben Sie sie, an der Zustellung ändert sich nichts.

FastAPI auf WordPlus Cloud

Mit WordPlus Cloud legt Ihre App ihre Räume einmal fest, jeweils mit einer Regel dafür, wer beitreten darf, und bindet einen einzigen Router ein:

from fastapi import FastAPI, Request
from starlette.middleware.sessions import SessionMiddleware
from wordplus_realtime import Realtime
import wordplus_realtime.fastapi as wr

realtime = Realtime.from_environment()   # WORDPLUS_SITE, WORDPLUS_SECRET, WORDPLUS_LOCAL_KEY

# The room of an order, order:1042: its customer may join, and the last 20 updates are kept.
realtime.room("order:{id}", lambda user, p: user is not None and owner_of(p["id"]) == user, history=20)

async def current_user(request: Request):
    return request.session.get("user")   # your sign-in

app = FastAPI()
app.add_middleware(SessionMiddleware, secret_key=SESSION_SECRET)
app.include_router(wr.router(realtime, current_user), prefix="/api/realtime")

Der Router beantwortet die Anfragen der Seiten, die einem Raum beitreten wollen, mit dem Nutzer, den Ihre App ermittelt, und liefert die Browserdateien des SDK aus, sodass Sie weder Node noch einen Bundler hinzufügen müssen. Eine Seite rendert realtime.head() für ihren Nutzer und tritt dann in einfachem JavaScript dem Raum bei:

import { connect } from '@wordplus/realtime';

const rt = await connect( window.wordplusRealtimeConfig );
rt.join( 'order:1042' ).on( 'updated', ( data ) => showStatus( data.status ) );

Und jeder Endpoint auf jedem beliebigen Worker sendet:

@app.post("/orders/{pk}/ship")
async def ship(pk: int):
    order = await mark_shipped(pk)
    await realtime.to(f"order:{pk}").aemit("updated", {"status": order.status})
    return {"ok": True}

Das läuft mit einem Worker genauso wie mit vierzig. Darüber hinaus:

• Ihre Regeln blockieren nie den Event Loop. Es sind einfache Funktionen, die in einem Worker-Thread laufen, sodass eine Datenbankabfrage darin kein Problem ist.
• Nichts geht bei einer Neuverbindung verloren. Ein Raum bewahrt seine letzten Events auf, sodass eine Seite, die zurückkommt, bekommt, was sie verpasst hat, in der richtigen Reihenfolge.
• Eine Verbindung für alle Tabs. Die Tabs eines Browsers teilen sich eine Verbindung, sodass eine Person mit drei offenen Tabs einmal zählt.
• Unterwegs versiegelt. Was Ihre App sendet, wird verschlüsselt übertragen, mit Schlüsseln, die nur Ihr Server und Ihre Nutzer haben.
• Mehr als Events. Online-Status, Anrufe, die in jedem Tab klingeln, Gruppenanrufe und Livestreams laufen über dieselbe Verbindung, mit fertigem Anrufbildschirm, Chat und Player aus Better Messages oder mit Ihrer eigenen Oberfläche auf derselben API.

Die FastAPI-Anleitung enthält die ganze Einrichtung, und die Live-Demo ist ein kleiner Shop, der darauf aufbaut und in dem eine Bestellung in einem anderen Fenster auf „versendet“ wechselt, während Sie klicken.

Für WordPlus Cloud for Apps gelten Preise nach gleichzeitigen Verbindungen, ab 1.000, monatlich oder jährlich abgerechnet. Sehen Sie, was sie für Apps leistet, oder werfen Sie einen Blick auf die Preise für Apps. Fragen zu Ihrer Integration? Schreiben Sie uns.