您现在的位置：程式碼之家 >> 編程語言 > >> Java 程式設計 >> Java 綜合教學

從零手寫 RPC 框架到 Dubbo 級服務治理：註冊發現、路由、負載均衡全實現

編輯：Java 綜合教學

從零手寫 RPC 框架到 Dubbo 級服務治理：註冊發現、路由、負載均衡全實現

RPC (Remote Procedure Call) 是分散式系統通訊基石。本文從 Socket 原始實現出發，層層遞進構建具備服務註冊發現、動態路由、負載均衡、容錯熔斷的企業級 RPC 框架核心模組，並對標 Dubbo 關鍵設計決策。

一、極簡 Socket RPC 原型 (教學用)

// 1. 定義介面
public interface HelloService { String sayHello(String name); }

// 2. 提供者
public class HelloServiceImpl implements HelloService {
    public String sayHello(String name) { return "Hello " + name; }
}

// 3. Server 端：阻塞 I/O + 反射調用
public class RpcServer {
    public static void main(String[] args) throws Exception {
        ServerSocket server = new ServerSocket(8080);
        while (true) {
            Socket socket = server.accept();
            new Thread(() -> {
                try (ObjectInput in = new ObjectInputStream(socket.getInputStream());
                     ObjectOutput out = new ObjectOutputStream(socket.getOutputStream())) {
                    
                    String className = in.readUTF();
                    String methodName = in.readUTF();
                    Class[] paramTypes = (Class[]) in.readObject();
                    Object[] args = (Object[]) in.readObject();
                    
                    Class clazz = Class.forName(className);
                    Object instance = clazz.getDeclaredConstructor().newInstance();
                    Method method = clazz.getMethod(methodName, paramTypes);
                    Object result = method.invoke(instance, args);
                    
                    out.writeObject(result);
                } catch (Exception e) { e.printStackTrace(); }
            }).start();
        }
    }
}

// 4. Client 端：動態代理隱藏網路細節
public class RpcClient {
    public static  T createProxy(Class interfaceClass, String host, int port) {
        return (T) Proxy.newProxyInstance(
            interfaceClass.getClassLoader(),
            new Class[]{interfaceClass},
            (proxy, method, args) -> {
                try (Socket socket = new Socket(host, port);
                     ObjectOutput out = new ObjectOutputStream(socket.getOutputStream());
                     ObjectInput in = new ObjectInputStream(socket.getInputStream())) {
                    
                    out.writeUTF(interfaceClass.getName());
                    out.writeUTF(method.getName());
                    out.writeObject(method.getParameterTypes());
                    out.writeObject(args);
                    out.flush();
                    return in.readObject();
                }
            }
        );
    }
    
    public static void main(String[] args) {
        HelloService service = createProxy(HelloService.class, "localhost", 8080);
        System.out.println(service.sayHello("RPC"));
    }
}

二、生產級核心模組架構

┌─────────────────────────────────────────────────────────────────┐
│                      RPC Framework Core                          │
├─────────────┬─────────────┬─────────────┬─────────────┬─────────┤
│  Protocol   │  Registry   │   Router    │ LoadBalance │ Filter  │
│  (編解碼/    │  (註冊發現/  │  (路由規則/  │ (負載均衡/  │ (攔截器 │
│   序列化)   │   健康檢查)  │   灰度發布)  │  容錯重試)  │  鏈)    │
└─────────────┴─────────────┴─────────────┴─────────────┴─────────┘
        ▲             ▲             ▲             ▲           ▲
        │             │             │             │           │
┌───────┴─────┐ ┌─────┴─────┐ ┌─────┴─────┐ ┌─────┴─────┐ ┌───┴────┐
│ Netty/Kryo  │ │ ZK/Nacos/ │ │ 規則引擎   │ │ 隨機/輪詢/ │ │ 監控/  │
│ /Protobuf   │ │ Etcd/Consul│ │ (Tag/版本/ │ │ 一致性Hash/│ │ 限流/  │
│             │ │            │ │  機房/權重)│ │  P2C/最小活躍│ │ 追蹤   │
└─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘ └────────┘

三、服務註冊與發現 (Registry)

1. 服務元數據模型

@Data @Builder
public class ServiceInstance {
    private String serviceName;      // 介面全限定名
    private String host;
    private int port;
    private Map metadata; // version, group, weight, zone, protocol...
    private long registerTime;
    private boolean healthy;
    private String instanceId;       // host:port:serviceName 唯一標識
}

2. Zookeeper 實現 (CP、強一致、Watch 機制)

public class ZkRegistry implements Registry {
    private static final String ROOT = "/rpc";
    private CuratorFramework client;
    
    @Override
    public void register(ServiceInstance instance) {
        String path = ROOT + "/" + instance.getServiceName() + "/" + instance.getInstanceId();
        // 臨時順序節點：Session 斷開自動刪除，實現健康檢查
        client.create().creatingParentsIfNeeded()
            .withMode(CreateMode.EPHEMERAL)
            .forPath(path, serialize(instance));
    }
    
    @Override
    public List discover(String serviceName) {
        String path = ROOT + "/" + serviceName;
        List children = client.getChildren().forPath(path);
        List instances = children.stream()
            .map(child -> deserialize(client.getData().forPath(path + "/" + child)))
            .filter(ServiceInstance::isHealthy)
            .collect(Collectors.toList());
        
        // 註冊 Watch：子節點變更時回調刷新本地緩存
        client.getChildren().usingWatcher((CuratorWatcher) event -> refreshCache(serviceName))
            .forPath(path);
        return instances;
    }
    
    // 本地緩存 + 定期全量同步雙重保障
}

3. Nacos 實現 (AP、最終一致、HTTP API、支援配置中心)

public class NacosRegistry implements Registry {
    private NamingService naming;
    
    @Override
    public void register(ServiceInstance instance) {
        naming.registerInstance(instance.getServiceName(), 
            instance.getHost(), instance.getPort(), instance.getMetadata());
        // 心跳由 Nacos 客戶端自動維護 (預設 5s)
    }
    
    @Override
    public List discover(String serviceName) {
        return naming.selectInstances(serviceName, true) // 僅健康實例
            .stream().map(this::convert).collect(Collectors.toList());
    }
    
    // 訂閱變更
    naming.subscribe(serviceName, event -> {
        if (event instanceof InstancesChangeEvent) {
            refreshLocalCache(serviceName);
        }
    });
}

四、動態路由

1. 路由規則模型

@Data
public class RouteRule {
    private String serviceName;
    private int priority;           // 優先級，數字越大越優先
    private Map condition; // 匹配條件：version=1.0.0, zone=shanghai, tag=canary
    private String targetCluster;   // 目標叢集/標籤
    private boolean enabled;
}

// 範例：灰度發布規則
RouteRule canaryRule = RouteRule.builder()
    .serviceName("com.demo.OrderService")
    .priority(100)
    .condition(Map.of("version", "2.0.0", "tag", "canary"))
    .targetCluster("canary-cluster")
    .build();

2. 路由引擎核心邏輯

public class RouterEngine {
    private final List rules = new CopyOnWriteArrayList<>();
    
    public List route(String serviceName, RpcInvocation invocation) {
        // 1. 獲取所有可用實例
        List allInstances = registry.discover(serviceName);
        
        // 2. 按優先級匹配規則
        for (RouteRule rule : rules) {
            if (!rule.isEnabled() || !rule.getServiceName().equals(serviceName)) continue;
            if (matchCondition(rule.getCondition(), invocation)) {
                // 3. 篩選目標實例
                List matched = allInstances.stream()
                    .filter(inst -> matchMetadata(rule.getTargetCluster(), inst))
                    .collect(Collectors.toList());
                if (!matched.isEmpty()) return matched; // 命中高優先級規則，直接返回
            }
        }
        return allInstances; // 無規則匹配，返回全量
    }
    
    private boolean matchCondition(Map condition, RpcInvocation inv) {
        return condition.entrySet().stream().allMatch(e -> {
            String value = inv.getAttachment(e.getKey()); // 從附件獲取：version, tag, zone...
            return e.getValue().equals(value);
        });
    }
}

五、負載均衡演算法實現

1. 介面定義

@FunctionalInterface
public interface LoadBalancer {
    ServiceInstance select(List instances, RpcInvocation invocation);
}

2. 核心演算法實現

// 2.1 隨機 (加權)
public class WeightedRandomLoadBalancer implements LoadBalancer {
    private final Random random = new Random();
    
    @Override
    public ServiceInstance select(List instances, RpcInvocation inv) {
        int totalWeight = instances.stream().mapToInt(i -> getWeight(i)).sum();
        int r = random.nextInt(totalWeight);
        for (ServiceInstance inst : instances) {
            r -= getWeight(inst);
            if (r < 0) return inst;
        }
        return instances.get(0);
    }
    private int getWeight(ServiceInstance i) { 
        return Integer.parseInt(i.getMetadata().getOrDefault("weight", "100")); 
    }
}

// 2.2 輪詢 (平滑加權 - Nginx 演算法)
public class SmoothRoundRobinLoadBalancer implements LoadBalancer {
    private final ConcurrentHashMap counters = new ConcurrentHashMap<>();
    
    @Override
    public ServiceInstance select(List instances, RpcInvocation inv) {
        String key = inv.getServiceName();
        AtomicInteger counter = counters.computeIfAbsent(key, k -> new AtomicInteger(0));
        int index = counter.getAndIncrement() % instances.size();
        return instances.get(index);
    }
}

// 2.3 一致性哈希 (參數級親和性)
public class ConsistentHashLoadBalancer implements LoadBalancer {
    private final ConsistentHash consistentHash = new ConsistentHash<>(160);
    
    @Override
    public ServiceInstance select(List instances, RpcInvocation inv) {
        // 根據調用參數 (如 userId) 計算哈希，保證同參數路由到同實例
        String key = inv.getParameter(0).toString(); // 假設第一參數為親和鍵
        return consistentHash.get(key);
    }
}

// 2.4 P2C (Power of Two Choices) - 最小活躍數 + 隨機
public class P2CLoadBalancer implements LoadBalancer {
    @Override
    public ServiceInstance select(List instances, RpcInvocation inv) {
        if (instances.size() <= 2) return instances.get(0);
        ServiceInstance i1 = instances.get(random.nextInt(instances.size()));
        ServiceInstance i2 = instances.get(random.nextInt(instances.size()));
        return getActiveCount(i1) < getActiveCount(i2) ? i1 : i2; // 活躍請求數少者優先
    }
}

六、容錯與熔斷 (Filter 鏈)

// 熔斷器狀態機
public class CircuitBreaker {
    private enum State { CLOSED, OPEN, HALF_OPEN }
    private State state = State.CLOSED;
    private int failureCount = 0;
    private int successCount = 0;
    private long lastFailureTime;
    
    public boolean allowRequest() {
        switch (state) {
            case CLOSED: return true;
            case OPEN: 
                if (System.currentTimeMillis() - lastFailureTime > timeout) {
                    state = State.HALF_OPEN;
                    return true;
                }
                return false;
            case HALF_OPEN: return true;
        }
        return false;
    }
    
    public void recordSuccess() {
        failureCount = 0;
        if (state == State.HALF_OPEN) {
            successCount++;
            if (successCount >= halfOpenSuccessThreshold) state = State.CLOSED;
        }
    }
    
    public void recordFailure() {
        failureCount++;
        lastFailureTime = System.currentTimeMillis();
        if (state == State.HALF_OPEN || failureCount >= failureThreshold) {
            state = State.OPEN;
        }
    }
}

// 重試 Filter
public class RetryFilter implements Filter {
    @Override
    public Result invoke(Invoker invoker, Invocation invocation) throws RpcException {
        int retries = invocation.getAttachment("retries", 2);
        for (int i = 0; i <= retries; i++) {
            try {
                return invoker.invoke(invocation);
            } catch (RpcException e) {
                if (i == retries || !isRetryable(e)) throw e;
                sleep(backoff(i));
            }
        }
        throw new RpcException("Retry exhausted");
    }
}

七、Dubbo 關鍵設計對標與啟發

領域	Dubbo 設計	手寫框架對標
配置中心	URL 統一配置模型、多配置源合並、動態刷新	引入 ConfigCenter 介面、支援 Nacos/Apollo、本地緩存
序列化	SPI 擴充：Hessian2/Kryo/Protobuf/FST/JSON	定義 Codec 介面、SPI 載入、預設 Kryo
通訊協定	Dubbo 協定 (單長連線、NIO、請求響應關聯)	Netty 實現自定義協議、RequestId 雙向映射
叢集容錯	Failover/Failfast/Failsafe/Forking/Broadcast	Cluster 介面 + Directory + LoadBalance 組合
服務治理	Admin Console、路由規則、動態配置、訪問控制	整合 Nacos 控制台、自建路由規則 DSL
監控指標	MetricsFilter 收集 RT/QPS/錯誤率、推送 Prometheus	Micrometer + Prometheus + Grafana

八、總結

從 Socket 原型到企業級 RPC，核心演進路徑是：網路模型升級 (BIO→NIO/Netty) → 服務發現解耦 (硬編碼→註冊中心) → 流量治理 (路由/負載/熔斷) → 可觀測性 (指標/追蹤/日誌) → 生態標準化 (SPI/協議/配置)。手寫框架過程中，重點攻克動態代理無感調用、註冊中心一致性與緩存、負載均衡演算法正確性、熔斷狀態機無鎖實現、序列化兼容性演進等硬骨頭。對標 Dubbo 設計哲學，以「微核+插件、URL 驅動、SPI 擴充」為架構指導，可構建出具備生產級可用性的 RPC 基礎設施。