Python实现长连接的多种方式解析

import requests

session = requests.Session()

for i in range(10):
    resp = session.get("http://localhost:8000/health")
    print(resp.status_code)

session.close()

HTTP → Upgrade → WebSocket

from fastapi import FastAPI, WebSocket
import asyncio

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(ws: WebSocket):
    await ws.accept()
    try:
        while True:
            data = await ws.receive_text()
            await ws.send_text(f"echo: {data}")
    except Exception:
        await ws.close()

import asyncio
import websockets

async def test_ws():
    async with websockets.connect("ws://localhost:8000/ws") as ws:
        await ws.send("hello")
        resp = await ws.recv()
        print(resp)

asyncio.run(test_ws())

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import time

app = FastAPI()

def event_generator():
    for i in range(5):
        yield f"data: message {i}\n\n"
        time.sleep(1)

@app.get("/sse")
def sse():
    return StreamingResponse(event_generator(),
                             media_type="text/event-stream")

import requests

resp = requests.get("http://localhost:8000/sse", stream=True)
for line in resp.iter_lines():
    if line:
        print(line.decode())

import asyncio

async def handle(reader, writer):
    while True:
        data = await reader.read(1024)
        if not data:
            break
        writer.write(data)
        await writer.drain()

asyncio.run(asyncio.start_server(handle, "0.0.0.0", 9000))

import asyncio

async def client():
    reader, writer = await asyncio.open_connection("127.0.0.1", 9000)
    writer.write(b"ping")
    await writer.drain()
    data = await reader.read(1024)
    print(data)
    writer.close()

asyncio.run(client())

import asyncio
import uvloop

asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

import redis

r = redis.Redis()

pubsub = r.pubsub()
pubsub.subscribe("events")

for msg in pubsub.listen():
    print(msg)

while True:
    ws.send("ping")
    time.sleep(30)

async def run(n=1000):
    tasks = [test_ws() for _ in range(n)]
    await asyncio.gather(*tasks)