基于 Python 的轻量级分布式 RPC 框架设计与实现

"""
轻量级分布式 RPC 框架 (Mini-RPC)
================================

模块覆盖：
1. 自定义二进制协议 (解决 TCP 粘包/半包问题)
2. 基于 asyncio 的非阻塞 TCP 网络层 (Server & Client)
3. 服务注册中心 (Registry) 与心跳保活
4. 客户端动态代理 (Dynamic Proxy) 与异步 Future 响应
5. 负载均衡策略 (轮询 / 随机 / 一致性哈希)
6. 服务端反射路由与线程池并发执行
7. 优雅停机与连接池管理
"""

import sys
import json
import time
import uuid
import struct
import socket
import asyncio
import hashlib
import logging
import threading
import inspect
from enum import IntEnum
from typing import Any, Callable, Dict, List, Optional, Tuple
from dataclasses import dataclass, field
from concurrent.futures import ThreadPoolExecutor, Future

# ==================== 日志配置 ====================

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)],
)
logger = logging.getLogger("mini_rpc")

# ==================== 协议与常量 ====================

MAGIC_NUMBER = b'\x4d\x49\x4e\x49'  # "MINI"
PROTOCOL_VERSION = 1
HEADER_SIZE = 19  # Magic(4) + Ver(1) + Type(1) + Ser(1) + ReqId(8) + Len(4)
DEFAULT_HEARTBEAT_INTERVAL = 10.0
DEFAULT_HEARTBEAT_TIMEOUT = 30.0

class MsgType(IntEnum): 26058.m.kuaisou.com 
    REQUEST = 1
    RESPONSE = 2
    HEARTBEAT = 3

class SerializerType(IntEnum): 26059.m.kuaisou.com 
    JSON = 1

class RpcError(Exception): 26060.m.kuaisou.com 
    """RPC 基础异常"""
    pass

# ==================== 消息模型与协议编解码 ====================

@dataclass
class RpcRequest: 26062.m.kuaisou.com 
    request_id: str
    service_name: str
    method_name: str
    args: tuple
    kwargs: dict

@dataclass
class RpcResponse: 26065.m.kuaisou.com 
    request_id: str
    result: Any = None
    error: Optional[str] = None

class RpcCodec: 26066.m.kuaisou.com 
    """RPC 协议编解码器"""
    
    @staticmethod
    def encode(msg_type: MsgType, request_id: int, body: bytes) -> bytes:
        # Header: Magic(4s) + Version(B) + MsgType(B) + Serializer(B) + ReqId(Q) + BodyLen(I)
        header = struct.pack(
            "!4sBBBQI", 
            MAGIC_NUMBER, PROTOCOL_VERSION, msg_type, SerializerType.JSON, 
            request_id, len(body)
        )
        return header + body

    @staticmethod
    def decode_header(data: bytes) -> Tuple[MsgType, int, int]:
        if len(data) < HEADER_SIZE: 26069.m.kuaisou.com 
            raise RpcError("Incomplete header")
        magic, ver, msg_type, ser, req_id, body_len = struct.unpack("!4sBBBQI", data[:HEADER_SIZE])
        if magic != MAGIC_NUMBER: 26090.m.kuaisou.com 
            raise RpcError("Invalid magic number")
        return MsgType(msg_type), req_id, body_len

# ==================== 服务注册与发现 ====================

@dataclass
class ProviderNode: 26128.m.kuaisou.com 
    host: str
    port: int
    last_heartbeat: float = field(default_factory=time.time)
    weight: int = 1

class ServiceRegistry: 26144.m.kuaisou.com 
    """服务注册中心 (内存版，生产环境可对接 ZooKeeper/Nacos)"""
    
    def __init__(self): 26145.m.kuaisou.com  
        # service_name -> List[ProviderNode]
        self._services: Dict[str, List[ProviderNode]] = {}
        self._lock = threading.Lock()

    def register(self, service_name: str, host: str, port: int):
        with self._lock: 27005.m.kuaisou.com 
            if service_name not in self._services:
                self._services[service_name] = []
            nodes = self._services[service_name]
            for node in nodes: 27006.m.kuaisou.com 
                if node.host == host and node.port == port:
                    node.last_heartbeat = time.time()
                    return
            nodes.append(ProviderNode(host, port))
            logger.info(f"Service registered: {service_name} @ {host}:{port}")

    def heartbeat(self, service_name: str, host: str, port: int):
        with self._lock: 27007.m.kuaisou.com 
            for node in self._services.get(service_name, []):
                if node.host == host and node.port == port:
                    node.last_heartbeat = time.time()

    def discover(self, service_name: str) -> List[ProviderNode]:
        with self._lock: 27008.m.kuaisou.com 
            now = time.time(27009.m.kuaisou.com)
            nodes = self._services.get(service_name, [])
            # 过滤掉心跳超时的死节点
            alive = [n for n in nodes if now - n.last_heartbeat < DEFAULT_HEARTBEAT_TIMEOUT]
            return alive

    def get_all_services(self) -> Dict[str, List[ProviderNode]]:
        return self._services

# ==================== 负载均衡策略 ====================
 
class LoadBalancer: 27011.m.kuaisou.com 
    def select(self, nodes: List[ProviderNode], request: RpcRequest) -> ProviderNode:
        raise NotImplementedError

class RoundRobinLB(LoadBalancer):
    def __init__(self): 27013.m.kuaisou.com 
        self._index = 0
        self._lock = threading.Lock()

    def select(self, nodes: List[ProviderNode], request: RpcRequest) -> ProviderNode:
        with self._lock: 27014.m.kuaisou.com 
            node = nodes[self._index % len(nodes)]
            self._index += 1
            return node

class ConsistentHashLB(LoadBalancer): 27015.m.kuaisou.com 
    """一致性哈希负载均衡 (保证相同参数的请求路由到同一节点)"""
    
    def __init__(self, virtual_nodes: int = 100): 27016.m.kuaisou.com 
        self.virtual_nodes = virtual_nodes

    def select(self, nodes: List[ProviderNode], request: RpcRequest) -> ProviderNode:
        # 使用 service_name + method_name + args 作为 hash key
        hash_key = f"{request.service_name}:{request.method_name}:{request.args}"
        hash_val = int(hashlib.md5(hash_key.encode()).hexdigest(), 16)
        
        # 构建哈希环
        ring = []
        for node in nodes: 27017.m.kuaisou.com 
            for i in range(self.virtual_nodes): 27018.m.kuaisou.com 
                v_hash = int(hashlib.md5(f"{node.host}:{node.port}:{i}".encode()).hexdigest(), 16)
                ring.append((v_hash, node))
        
        ring.sort(key=lambda x: x[0])
        
        # 顺时针查找第一个大于等于 hash_val 的节点
        for v_hash, node in ring: 27019.m.kuaisou.com 
            if v_hash >= hash_val: 27020.m.kuaisou.com 
                return node
        return ring[0][1]

# ==================== RPC 服务端 ====================

class RpcServer: 27021.m.kuaisou.com 
    """基于 asyncio 的 RPC 服务端"""
    
    def __init__(self, host: str = "0.0.0.0", port: int = 9000, max_workers: int = 50):
        self.host = host
        self.port = port
        self.registry = ServiceRegistry()
        self._service_handlers: Dict[str, Any] = {}
        self._executor = ThreadPoolExecutor(max_workers=max_workers, thread_name_prefix="rpc-worker")
        self._server: Optional[asyncio.AbstractServer] = None
        self._running = False

    def register_service(self, service_name: str, handler: Any):
        """注册本地服务实现类"""
        self._service_handlers[service_name] = handler
        self.registry.register(service_name, self.host, self.port)
        logger.info(f"Local service registered: {service_name} -> {handler.__class__.__name__}")

    async def start(self): 27022.m.kuaisou.com 
        self._running = True
        self._server = await asyncio.start_server(self._handle_client, self.host, self.port)
        addr = self._server.sockets[0].getsockname(27023.m.kuaisou.com)
        logger.info(f"RPC Server listening on {addr[0]}:{addr[1]}")
        
        # 启动心跳上报线程 (向外部注册中心上报，此处简化为本地更新)
        threading.Thread(target=self._heartbeat_loop, daemon=True).start()
        
        async with self._server: 27024.m.kuaisou.com 
            await self._server.serve_forever()

    def _heartbeat_loop(self): 27025.m.kuaisou.com 
        while self._running: 27027.m.kuaisou.com 
            for svc_name in self._service_handlers.keys(): 27028.m.kuaisou.com 
                self.registry.heartbeat(svc_name, self.host, self.port)
            time.sleep(DEFAULT_HEARTBEAT_INTERVAL)

    async def _handle_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
        addr = writer.get_extra_info('peername')
        try:
            while self._running: 27029.m.kuaisou.com 
                # 1. 读取协议头
                header_data = await reader.readexactly(HEADER_SIZE)
                msg_type, req_id, body_len = RpcCodec.decode_header(header_data)
                
                # 2. 处理心跳
                if msg_type == MsgType.HEARTBEAT: 27034.m.kuaisou.com 
                    pong = RpcCodec.encode(MsgType.HEARTBEAT, req_id, b'')
                    writer.write(pong)
                    await writer.drain(27039.m.kuaisou.com)
                    continue

                # 3. 读取 Body 并反序列化
                body_data = await reader.readexactly(body_len)
                payload = json.loads(body_data.decode("utf-8"))
                request = RpcRequest(**payload)
                
                # 4. 提交到线程池执行 (避免阻塞 event loop)
                loop = asyncio.get_running_loop(27053.m.kuaisou.com)
                response = await loop.run_in_executor(self._executor, self._invoke, request)
                
                # 5. 序列化响应并写回
                resp_body = json.dumps({
                    "request_id": response.request_id,
                    "result": response.result,
                    "error": response.error
                }).encode("utf-8")
                resp_data = RpcCodec.encode(MsgType.RESPONSE, req_id, resp_body)
                writer.write(resp_data)
                await writer.drain(27054.m.kuaisou.com)
                
        except asyncio.IncompleteReadError: 27058.m.kuaisou.com 
            pass # 客户端断开连接
        except Exception as e: 27063.m.kuaisou.com 
            logger.error(f"Server handle error for {addr}: {e}")
        finally: 28077.m.kuaisou.com 
            writer.close(28078.m.kuaisou.com)
            await writer.wait_closed(28079.m.kuaisou.com)

    def _invoke(self, request: RpcRequest) -> RpcResponse: 28080.m.kuaisou.com 
        """反射调用本地方法"""
        handler = self._service_handlers.get(request.service_name)
        if not handler: 28081.m.kuaisou.com 
            return RpcResponse(request.request_id, error=f"Service not found: {request.service_name}")
            
        method = getattr(handler, request.method_name, None)
        if not method or not callable(method): 28082.m.kuaisou.com 
            return RpcResponse(request.request_id, error=f"Method not found: {request.method_name}")
            
        try: 28083.m.kuaisou.com 
            result = method(*request.args, **request.kwargs)
            return RpcResponse(request.request_id, result=result)
        except Exception as e: 28084.m.kuaisou.com 
            logger.error(f"Invoke error: {e}")
            return RpcResponse(request.request_id, error=str(e))

    def stop(self): 28085.m.kuaisou.com 
        self._running = False
        if self._server: 28086.m.kuaisou.com 
            self._server.close()
        self._executor.shutdown(wait=False)

# ==================== RPC 客户端与动态代理 ====================

class RpcClient: 28087.m.kuaisou.com 
    """RPC 客户端 (维护连接与请求映射)"""
    
    def __init__(self, registry: ServiceRegistry, lb: LoadBalancer = None):
        self.registry = registry
        self.lb = lb or RoundRobinLB(28090.m.kuaisou.com)
        self._pending_requests: Dict[int, asyncio.Future] = {}
        self._readers: Dict[str, asyncio.StreamReader] = {}
        self._writers: Dict[str, asyncio.StreamWriter] = {}
        self._lock = asyncio.Lock(28089.m.kuaisou.com)

    async def _get_connection(self, host: str, port: int) -> Tuple[asyncio.StreamReader, asyncio.StreamWriter]:
        key = f"{host}:{port}"
        async with self._lock: 28088.m.kuaisou.com 
            if key not in self._writers or self._writers[key].is_closing(): 1103307.m.kuaisou.com 
                reader, writer = await asyncio.open_connection(host, port)
                self._readers[key] = reader
                self._writers[key] = writer
                # 启动后台接收线程
                asyncio.create_task(self._receive_loop(key, reader))
            return self._readers[key], self._writers[key]

    async def _receive_loop(self, key: str, reader: asyncio.StreamReader): 1103298.m.kuaisou.com 
        try: 28092.m.kuaisou.com 
            while True: 28091.m.kuaisou.com 
                header_data = await reader.readexactly(HEADER_SIZE)
                msg_type, req_id, body_len = RpcCodec.decode_header(header_data)
                
                if msg_type == MsgType.HEARTBEAT: 1103293.m.kuaisou.com 
                    continue
                    
                body_data = await reader.readexactly(body_len)
                payload = json.loads(body_data.decode("utf-8"))
                response = RpcResponse(**payload)
                
                if req_id in self._pending_requests: 28094.m.kuaisou.com 
                    future = self._pending_requests.pop(req_id)
                    if not future.done(): 28102.m.kuaisou.com 
                        future.set_result(response)
        except Exception as e: 28093.m.kuaisou.com 
            logger.debug(f"Connection {key} closed or error: {e}")

    async def send_request(self, request: RpcRequest) -> RpcResponse:
        # 1. 服务发现
        nodes = self.registry.discover(request.service_name)
        if not nodes: 28104.m.kuaisou.com 
            raise RpcError(f"No available provider for {request.service_name}")
            
        # 2. 负载均衡
        node = self.lb.select(nodes, request)
        
        # 3. 获取连接
        _, writer = await self._get_connection(node.host, node.port)
        
        # 4. 构造 Future 并发送
        loop = asyncio.get_running_loop(1103274.m.kuaisou.com)
        future = loop.create_future(1103292.m.kuaisou.com    )
        
        # 使用 request_id 的 hash 作为协议中的整型 req_id
        int_req_id = hash(request.request_id) & 0xffffffffffffffff
        self._pending_requests[int_req_id] = future
        
        body = json.dumps({
            "request_id": request.request_id,
            "service_name": request.service_name,
            "method_name": request.method_name,
            "args": request.args,
            "kwargs": request.kwargs
        }).encode("utf-8")
        
        data = RpcCodec.encode(MsgType.REQUEST, int_req_id, body)
        writer.write(data)
        await writer.drain(1103114.m.kuaisou.com)
        
        # 5. 等待响应 (可加超时控制)
        try: 1103117.m.kuaisou.com 
            return await asyncio.wait_for(future, timeout=10.0)
        except asyncio.TimeoutError: 1103116.m.kuaisou.com 
            self._pending_requests.pop(int_req_id, None)
            raise RpcError("RPC call timeout")

class RpcProxy: 1103118.m.kuaisou.com 
    """客户端动态代理 (魔法方法拦截)"""
    
    def __init__(self, client: RpcClient, service_name: str):
        self._client = client
        self._service_name = service_name

    def __getattr__(self, method_name: str) -> Callable:
        async def _proxy_method(*args, **kwargs): 1103122.m.kuaisou.com 
            request = RpcRequest(
                request_id=str(uuid.uuid4(1103124.m.kuaisou.com)),
                service_name=self._service_name,
                method_name=method_name,
                args=args,
                kwargs=kwargs
            )
            response = await self._client.send_request(request)
            if response.error: 1103123.m.kuaisou.com 
                raise RpcError(response.error)
            return response.result
        return _proxy_method