10 - 现代网络协议¶

# QUIC使用连接ID而非IP:Port标识连接
# 网络切换时（WiFi → 4G）连接不中断

class QUICConnection:
    def __init__(self):
        self.connection_id = generate_random_cid()  # 变长连接ID（0-20字节，RFC 9000）
        self.paths = []  # 支持多路径

    def migrate(self, new_local_address):
        """网络切换时迁移连接"""
        # 发送PATH_CHALLENGE验证新路径
        # 保持connection_id不变
        # 连接继续，不中断！
        pass

from aioquic.asyncio import QuicConnectionProtocol, serve
from aioquic.quic.configuration import QuicConfiguration
from aioquic.quic.events import StreamDataReceived
from aioquic.h3.connection import H3Connection
from aioquic.h3.events import DataReceived, RequestReceived

class Http3Server(QuicConnectionProtocol):
    def __init__(self, *args, **kwargs):  # *args接收任意位置参数；**kwargs接收任意关键字参数
        super().__init__(*args, **kwargs)  # super()调用父类方法
        self._http = None

    def quic_event_received(self, event):
        if isinstance(event, StreamDataReceived):
            if self._http is None:
                self._http = H3Connection(self._quic)

            for http_event in self._http.handle_event(event):
                self._handle_http_event(http_event)

    def _handle_http_event(self, event):
        if isinstance(event, RequestReceived):
            # 处理HTTP/3请求
            self._http.send_response(
                stream_id=event.stream_id,
                headers=[(b":status", b"200")],
                data=b"Hello HTTP/3!"
            )

# 启动HTTP/3服务器
configuration = QuicConfiguration(
    alpn_protocols=["h3"],
    is_client=False,
    max_datagram_frame_size=65536
)

configuration.load_cert_chain("cert.pem", "key.pem")

loop = asyncio.get_event_loop()
loop.run_until_complete(
    serve(
        host="0.0.0.0",
        port=4433,
        configuration=configuration,
        create_protocol=Http3Server
    )
)

import aioquic
from aioquic.asyncio import connect
from aioquic.quic.configuration import QuicConfiguration
from aioquic.h3.connection import H3Connection
from aioquic.h3.events import ResponseReceived

async def http3_request(host, port, path):
    """发起HTTP/3请求"""
    configuration = QuicConfiguration(
        alpn_protocols=["h3"],
        is_client=True
    )

    async with connect(host, port, configuration=configuration) as client:
        http = H3Connection(client)

        # 发送请求
        stream_id = client.get_next_available_stream_id()
        http.send_request(
            stream_id=stream_id,
            headers=[
                (b":method", b"GET"),
                (b":scheme", b"https"),
                (b":authority", host.encode()),
                (b":path", path.encode()),
            ]
        )

        # 等待响应
        while True:
            event = await client.wait_for_event()
            for http_event in http.handle_event(event):
                if isinstance(http_event, ResponseReceived):  # isinstance检查对象类型
                    print(f"Status: {http_event.headers}")
                    print(f"Data: {http_event.data}")
                    return

# 使用
asyncio.run(http3_request("cloudflare-quic.com", 443, "/"))

# server.py
import grpc
from concurrent import futures
import service_pb2
import service_pb2_grpc

class ModelServiceServicer(service_pb2_grpc.ModelServiceServicer):
    """实现ModelService服务"""

    def Predict(self, request, context):
        """简单RPC：单次预测"""
        print(f"收到预测请求: model={request.model_id}")

        # 执行模型推理
        result = self._run_inference(request.input_data)

        return service_pb2.PredictResponse(
            output_data=result,
            confidence=0.95,
            inference_time_ms=100
        )

    def StreamPredict(self, request, context):
        """服务端流式：流式返回结果"""
        for chunk in self._stream_inference(request.input_data):
            yield service_pb2.PredictResponse(
                output_data=chunk,
                confidence=0.9
            )

    def BatchPredict(self, request_iterator, context):
        """客户端流式：批量处理"""
        results = []
        for request in request_iterator:
            result = self._run_inference(request.input_data)
            results.append(result)

        return service_pb2.BatchResponse(
            results=results,
            total_count=len(results)
        )

    def Chat(self, request_iterator, context):
        """双向流式：实时对话"""
        for message in request_iterator:
            # 处理用户消息
            response = self._generate_response(message)

            yield service_pb2.ChatMessage(
                role="assistant",
                content=response,
                timestamp=int(time.time())
            )

def serve():
    # 创建gRPC服务器
    server = grpc.server(
        futures.ThreadPoolExecutor(max_workers=10),
        options=[
            ('grpc.max_send_message_length', 50 * 1024 * 1024),
            ('grpc.max_receive_message_length', 50 * 1024 * 1024),
        ]
    )

    # 注册服务
    service_pb2_grpc.add_ModelServiceServicer_to_server(
        ModelServiceServicer(), server
    )

    # 绑定端口（支持TLS）
    server.add_secure_port(
        '[::]:50051',  # 切片操作：[start:end:step]提取子序列
        grpc.ssl_server_credentials(
            ((private_key, certificate_chain),)
        )
    )

    server.start()
    print("gRPC服务器启动在端口 50051")
    server.wait_for_termination()

if __name__ == '__main__':
    serve()

# client.py
import grpc
import service_pb2
import service_pb2_grpc

class ModelClient:
    def __init__(self, target='localhost:50051'):
        # 创建通道（支持负载均衡）
        self.channel = grpc.insecure_channel(
            target,
            options=[
                ('grpc.lb_policy_name', 'round_robin'),
                ('grpc.keepalive_time_ms', 10000),
            ]
        )
        self.stub = service_pb2_grpc.ModelServiceStub(self.channel)

    def predict(self, model_id, input_data):
        """简单RPC调用"""
        request = service_pb2.PredictRequest(
            model_id=model_id,
            input_data=input_data
        )

        try:
            response = self.stub.Predict(request, timeout=30)
            return {
                'output': response.output_data,
                'confidence': response.confidence,
                'time_ms': response.inference_time_ms
            }
        except grpc.RpcError as e:
            print(f"RPC错误: {e.code()}: {e.details()}")
            raise

    def stream_predict(self, model_id, input_data):
        """服务端流式调用"""
        request = service_pb2.PredictRequest(
            model_id=model_id,
            input_data=input_data
        )

        for response in self.stub.StreamPredict(request):
            yield {
                'chunk': response.output_data,
                'confidence': response.confidence
            }

    def chat_stream(self):
        """双向流式对话"""
        def message_generator():
            while True:
                user_input = input("你: ")
                if user_input == 'quit':
                    break
                yield service_pb2.ChatMessage(
                    role="user",
                    content=user_input
                )

        responses = self.stub.Chat(message_generator())
        for response in responses:
            print(f"AI: {response.content}")

# 使用示例
client = ModelClient('localhost:50051')

# 简单调用
result = client.predict('gpt-4', b'Hello')

# 流式调用
for chunk in client.stream_predict('gpt-4', b'Long text...'):
    print(chunk)

import time
import grpc
from concurrent import futures

# 认证拦截器
class AuthInterceptor(grpc.ServerInterceptor):
    def __init__(self, validator):
        self.validator = validator

    def intercept_service(self, continuation, handler_call_details):
        """
        intercept_service 的正确用法：
        - continuation(handler_call_details) → 返回 RpcMethodHandler（非请求处理）
        - 返回值必须是 RpcMethodHandler 或 None
        """
        # 1. 先调用 continuation 获取原始 handler
        handler = continuation(handler_call_details)
        if handler is None:
            return None

        # 2. 包装原始处理函数，添加认证逻辑
        def _auth_wrapper(behavior):
            def wrapper(request_or_iterator, context):
                metadata = dict(context.invocation_metadata())
                token = metadata.get('authorization', '')
                if not self.validator(token):
                    context.abort(grpc.StatusCode.UNAUTHENTICATED, '无效的认证令牌')
                return behavior(request_or_iterator, context)
            return wrapper

        # 3. 根据 handler 类型包装（此处以 unary_unary 为例）
        if handler.unary_unary:
            return grpc.unary_unary_rpc_method_handler(
                _auth_wrapper(handler.unary_unary),
                request_deserializer=handler.request_deserializer,
                response_serializer=handler.response_serializer
            )
        # 其他类型（stream_unary, unary_stream, stream_stream）类似包装
        return handler

# 日志拦截器
class LoggingInterceptor(grpc.ServerInterceptor):
    def intercept_service(self, continuation, handler_call_details):
        method = handler_call_details.method
        handler = continuation(handler_call_details)
        if handler is None:
            return None

        def _log_wrapper(behavior):
            def wrapper(request_or_iterator, context):
                start = time.time()
                response = behavior(request_or_iterator, context)
                duration = time.time() - start
                print(f"[{method}] 耗时: {duration:.3f}s")
                return response
            return wrapper

        if handler.unary_unary:
            return grpc.unary_unary_rpc_method_handler(
                _log_wrapper(handler.unary_unary),
                request_deserializer=handler.request_deserializer,
                response_serializer=handler.response_serializer
            )
        return handler

# 注册拦截器
server = grpc.server(
    futures.ThreadPoolExecutor(max_workers=10),  # 线程池/多线程：并发执行任务
    interceptors=[
        AuthInterceptor(validate_token),
        LoggingInterceptor()
    ]
)

# 健康检查服务
from grpc_health.v1 import health_pb2, health_pb2_grpc

class HealthServicer(health_pb2_grpc.HealthServicer):
    def __init__(self):
        self.status = health_pb2.HealthCheckResponse.SERVING

    def Check(self, request, context):
        return health_pb2.HealthCheckResponse(status=self.status)

    def Watch(self, request, context):
        # 流式健康状态
        while True:
            yield health_pb2.HealthCheckResponse(status=self.status)  # yield生成器：惰性产出值，节省内存
            time.sleep(5)

# 客户端负载均衡
from grpc import load_balancer

channel = grpc.insecure_channel(
    'dns:///ai-service.example.com:50051',
    options=[
        ('grpc.lb_policy_name', 'round_robin'),
        ('grpc.service_config', json.dumps({
            'loadBalancingConfig': [{'round_robin': {}}],
            'healthCheckConfig': {'serviceName': 'ai_service.ModelService'}
        }))
    ]
)

import asyncio
import websockets
import json
from collections import defaultdict

class ChatServer:
    def __init__(self):
        # 房间管理
        self.rooms = defaultdict(set)
        # 用户连接映射
        self.users = {}

    async def register(self, websocket, room_id, username):
        """注册用户到房间"""
        self.rooms[room_id].add(websocket)
        self.users[websocket] = {
            'username': username,
            'room': room_id
        }

        # 广播用户加入
        await self.broadcast(room_id, {
            'type': 'system',
            'message': f'{username} 加入了房间'
        })

    async def unregister(self, websocket):
        """移除用户"""
        user = self.users.pop(websocket, None)
        if user:
            room_id = user['room']
            self.rooms[room_id].discard(websocket)

            await self.broadcast(room_id, {
                'type': 'system',
                'message': f"{user['username']} 离开了房间"
            })

    async def broadcast(self, room_id, message):
        """广播消息到房间"""
        if room_id not in self.rooms:
            return

        message_str = json.dumps(message)

        # 并发发送给所有客户端
        await asyncio.gather(
            *[ws.send(message_str) for ws in self.rooms[room_id]],
            return_exceptions=True
        )

    async def handler(self, websocket, path):
        """处理WebSocket连接"""
        try:
            # 等待初始化消息
            init_msg = await websocket.recv()
            data = json.loads(init_msg)

            room_id = data.get('room', 'default')
            username = data.get('username', 'Anonymous')

            await self.register(websocket, room_id, username)

            # 消息循环
            async for message in websocket:
                try:
                    data = json.loads(message)

                    # 广播用户消息
                    await self.broadcast(room_id, {
                        'type': 'message',
                        'username': username,
                        'content': data.get('content'),
                        'timestamp': asyncio.get_event_loop().time()
                    })

                except json.JSONDecodeError:
                    await websocket.send(json.dumps({
                        'type': 'error',
                        'message': '无效的JSON格式'
                    }))

        except websockets.exceptions.ConnectionClosed:
            pass
        finally:
            await self.unregister(websocket)

# 启动服务器
server = ChatServer()
start_server = websockets.serve(server.handler, "localhost", 8765)

asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()

import asyncio
import websockets
import json
import aioconsole

async def chat_client():
    """WebSocket聊天客户端"""
    uri = "ws://localhost:8765"

    async with websockets.connect(uri) as websocket:
        # 发送初始化信息
        username = input("请输入用户名: ")
        await websocket.send(json.dumps({
            'username': username,
            'room': 'general'
        }))

        # 启动接收任务
        receive_task = asyncio.create_task(receive_messages(websocket))

        # 发送消息循环
        while True:
            message = await aioconsole.ainput()
            if message == '/quit':
                break

            await websocket.send(json.dumps({
                'content': message
            }))

        receive_task.cancel()

async def receive_messages(websocket):
    """接收服务器消息"""
    try:
        async for message in websocket:
            data = json.loads(message)

            if data['type'] == 'system':
                print(f"[系统] {data['message']}")
            elif data['type'] == 'message':
                print(f"[{data['username']}] {data['content']}")

    except asyncio.CancelledError:
        pass

# 运行
asyncio.run(chat_client())  # asyncio.run()启动异步事件循环

import asyncio

class HeartbeatManager:
    """WebSocket心跳管理"""

    def __init__(self, interval=30, timeout=10):
        self.interval = interval  # 心跳间隔
        self.timeout = timeout    # 超时时间
        self.last_pong = {}

    async def start_heartbeat(self, websocket):
        """启动心跳"""
        self.last_pong[websocket] = asyncio.get_event_loop().time()

        try:
            while True:
                await asyncio.sleep(self.interval)

                # 发送ping
                await websocket.ping()

                # 检查上次pong时间
                last_pong = self.last_pong.get(websocket, 0)
                if asyncio.get_event_loop().time() - last_pong > self.timeout:
                    print("心跳超时，关闭连接")
                    await websocket.close()
                    break

        except websockets.exceptions.ConnectionClosed:
            pass

    def on_pong(self, websocket):
        """收到pong响应"""
        self.last_pong[websocket] = asyncio.get_event_loop().time()

import redis.asyncio as aioredis

class DistributedChatServer:
    """分布式WebSocket服务器"""

    def __init__(self):
        self.redis = None
        self.local_clients = defaultdict(set)  # defaultdict带默认值的字典，避免KeyError

    async def init_redis(self):  # async def定义异步函数；用await调用
        """初始化Redis连接"""
        self.redis = await aioredis.from_url("redis://localhost")  # await等待异步操作完成

        # 启动订阅任务
        asyncio.create_task(self._subscribe_messages())

    async def _subscribe_messages(self):
        """订阅Redis频道"""
        pubsub = self.redis.pubsub()
        await pubsub.subscribe('chat:messages')

        async for message in pubsub.listen():
            if message['type'] == 'message':
                data = json.loads(message['data'])  # json.loads将JSON字符串转为Python对象
                room_id = data['room']

                # 发送给本地该房间的客户端
                for ws in self.local_clients[room_id]:
                    try:  # try/except捕获异常
                        await ws.send(message['data'])
                    except:
                        pass

    async def broadcast(self, room_id, message):
        """广播到整个集群"""
        message['room'] = room_id

        # 发布到Redis
        await self.redis.publish(
            'chat:messages',
            json.dumps(message)  # json.dumps将Python对象转为JSON字符串
        )

# 目标：创建一个AI模型推理微服务
# 1. 定义.proto接口
# 2. 实现Python服务端
# 3. 实现Go客户端
# 4. 添加认证和日志
# 5. 部署到Docker

# 目标：使用WebSocket构建类似Google Docs的协作编辑器
# 功能：
# - 多用户实时编辑
# - 光标位置同步
# - 操作冲突解决（OT算法）
# - 版本历史