從零打造輕量級分散式 RPC 框架：Spring + Netty + Protostuff + ZooKeeper 完整實戰教學

前言：為什麼要手寫 RPC 框架？

RPC (Remote Procedure Call) 讓「呼叫遠端服務如同呼叫本地方法」成為可能。雖然生產環境首選 Dubbo/gRPC，但親手實作一個基於 TCP + NIO + 高效序列化 + 服務註冊發現 的輕量級 RPC 框架，能深度理解：網路傳輸、序列化協議、服務治理、代理模式、Netty 事件模型等分散式系統核心機制。本教學將分 10 步驟完整構建可運行框架，包含所有核心代碼與 Maven 依賴。

技術選型與架構圖

架構流程

1. Provider 啟動 → Spring 掃描 @RpcService Bean → Netty Server 綁定端口 → 向 ZooKeeper 註冊臨時節點 (/registry/data/服務名_地址)
2. Consumer 啟動 → Spring 建立 RpcProxy → ZooKeeper 監聽 /registry/data 子節點變化 → 本地緩存服務列表
3. Consumer 呼叫代理方法 → RpcProxy 組裝 RpcRequest → Netty Client 發送 → Provider RpcHandler 解碼 → FastClass 快速調用實現類 → 封裝 RpcResponse 返回

十步驟完整實作

步驟 1：定義服務介面 (獨立 API 模組)

public interface HelloService { String hello(String name); }

步驟 2：實作服務提供者

@RpcService(HelloService.class) // 標記為 RPC 服務，指定介面 public class HelloServiceImpl implements HelloService { @Override public String hello(String name) { return "Hello! " + name; } }

步驟 3：@RpcService 註解定義 (含 Spring @Component)

@Target(ElementType.TYPE) @Retention(RetentionPolicy.RUNTIME) @Component // 可被 Spring 掃描 public @interface RpcService { Class<?> value(); // 指定遠端介面 }

步驟 4：Provider 端 Spring 配置 (spring.xml)

<beans ...> <context:component-scan base-package="com.example.rpc.server"/> <context:property-placeholder location="classpath:config.properties"/> <bean id="serviceRegistry" class="com.example.rpc.registry.ServiceRegistry"> <constructor-arg name="registryAddress" value="${registry.address}"/> </bean> <bean id="rpcServer" class="com.example.rpc.server.RpcServer"> <constructor-arg name="serverAddress" value="${server.address}"/> <constructor-arg name="serviceRegistry" ref="serviceRegistry"/> </bean> </beans>

config.properties

registry.address=127.0.0.1:2181 server.address=127.0.0.1:8000

步驟 5：啟動引導類

public class RpcBootstrap { public static void main(String[] args) { new ClassPathXmlApplicationContext("spring.xml"); } }

步驟 6：服務註冊 (ServiceRegistry - ZooKeeper)

public class ServiceRegistry { private static final Logger LOG = LoggerFactory.getLogger(ServiceRegistry.class); private final CountDownLatch latch = new CountDownLatch(1); private final String registryAddress; public ServiceRegistry(String registryAddress) { this.registryAddress = registryAddress; } public void register(String serverAddress) { if (serverAddress == null) return; ZooKeeper zk = connect(); if (zk != null) createNode(zk, serverAddress); } private ZooKeeper connect() { try { ZooKeeper zk = new ZooKeeper(registryAddress, Constant.ZK_SESSION_TIMEOUT, event -> { if (event.getState() == KeeperState.SyncConnected) latch.countDown(); }); latch.await(); return zk; } catch (Exception e) { LOG.error("ZK connect failed", e); return null; } } private void createNode(ZooKeeper zk, String data) { try { byte[] bytes = data.getBytes(); String path = zk.create(Constant.ZK_DATA_PATH, bytes, Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); LOG.debug("Registered: {} => {}", path, data); } catch (Exception e) { LOG.error("Create node failed", e); } } }

Constant 常量

public interface Constant { int ZK_SESSION_TIMEOUT = 5000; String ZK_REGISTRY_PATH = "/registry"; String ZK_DATA_PATH = ZK_REGISTRY_PATH + "/data"; }

注意：需先在 ZK CLI 建立永久節點 /registry。

步驟 7：RPC 服務端核心 (RpcServer - Netty)

public class RpcServer implements ApplicationContextAware, InitializingBean { private final String serverAddress; private final ServiceRegistry serviceRegistry; private final Map<String, Object> handlerMap = new HashMap<>(); // 介面名 -> 實現 Bean public RpcServer(String serverAddress, ServiceRegistry registry) { this.serverAddress = serverAddress; this.serviceRegistry = registry; } @Override public void setApplicationContext(ApplicationContext ctx) { Map<String, Object> beans = ctx.getBeansWithAnnotation(RpcService.class); for (Object bean : beans.values()) { String interfaceName = bean.getClass().getAnnotation(RpcService.class).value().getName(); handlerMap.put(interfaceName, bean); } } @Override public void afterPropertiesSet() { EventLoopGroup boss = new NioEventLoopGroup(); EventLoopGroup worker = new NioEventLoopGroup(); try { ServerBootstrap b = new ServerBootstrap(); b.group(boss, worker).channel(NioServerSocketChannel.class) .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) { ch.pipeline() .addLast(new RpcDecoder(RpcRequest.class)) // 解碼請求 .addLast(new RpcEncoder(RpcResponse.class)) // 編碼響應 .addLast(new RpcHandler(handlerMap)); // 處理邏輯 } }) .option(ChannelOption.SO_BACKLOG, 128) .childOption(ChannelOption.SO_KEEPALIVE, true); String[] parts = serverAddress.split(":"); ChannelFuture f = b.bind(parts[0], Integer.parseInt(parts[1])).sync(); LOG.info("RPC Server started on {}", serverAddress); if (serviceRegistry != null) serviceRegistry.register(serverAddress); f.channel().closeFuture().sync(); } catch (Exception e) { LOG.error("Server error", e); } finally { worker.shutdownGracefully(); boss.shutdownGracefully(); } } }

步驟 8：核心傳輸對象與編解碼器

// RpcRequest.java public class RpcRequest { private String requestId; private String className; private String methodName; private Class<?>[] parameterTypes; private Object[] parameters; // getter/setter } // RpcResponse.java public class RpcResponse { private String requestId; private Throwable error; private Object result; public boolean isError() { return error != null; } // getter/setter }

// RpcDecoder.java (Netty ByteToMessageDecoder) public class RpcDecoder extends ByteToMessageDecoder { private final Class<?> clazz; public RpcDecoder(Class<?> clazz) { this.clazz = clazz; } @Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) { if (in.readableBytes() < 4) return; in.markReaderIndex(); int len = in.readInt(); if (len < 0) { ctx.close(); return; } if (in.readableBytes() < len) { in.resetReaderIndex(); return; } byte[] data = new byte[len]; in.readBytes(data); Object obj = SerializationUtil.deserialize(data, clazz); out.add(obj); } } // RpcEncoder.java (Netty MessageToByteEncoder) public class RpcEncoder extends MessageToByteEncoder<Object> { private final Class<?> clazz; public RpcEncoder(Class<?> clazz) { this.clazz = clazz; } @Override protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) { if (clazz.isInstance(msg)) { byte[] data = SerializationUtil.serialize(msg); out.writeInt(data.length); out.writeBytes(data); } } }

步驟 9：序列化工具類 (Protostuff + Objenesis)

public class SerializationUtil { private static final Map<Class<?>, Schema<?>> SCHEMA_CACHE = new ConcurrentHashMap<>(); private static final Objenesis OBJENESIS = new ObjenesisStd(true); private SerializationUtil() {} @SuppressWarnings("unchecked") private static <T> Schema<T> getSchema(Class<T> cls) { Schema<T> schema = (Schema<T>) SCHEMA_CACHE.get(cls); if (schema == null) { schema = RuntimeSchema.createFrom(cls); if (schema != null) SCHEMA_CACHE.put(cls, schema); } return schema; } public static <T> byte[] serialize(T obj) { Class<T> cls = (Class<T>) obj.getClass(); LinkedBuffer buffer = LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE); try { Schema<T> schema = getSchema(cls); return ProtostuffIOUtil.toByteArray(obj, schema, buffer); } finally { buffer.clear(); } } public static <T> T deserialize(byte[] data, Class<T> cls) { try { T obj = OBJENESIS.newInstance(cls); Schema<T> schema = getSchema(cls); ProtostuffIOUtil.mergeFrom(data, obj, schema); return obj; } catch (Exception e) { throw new IllegalStateException(e); } } }

步驟 10：請求處理器 (RpcHandler - CGLib FastClass)

public class RpcHandler extends SimpleChannelInboundHandler<RpcRequest> { private final Map<String, Object> handlerMap; public RpcHandler(Map<String, Object> handlerMap) { this.handlerMap = handlerMap; } @Override protected void channelRead0(ChannelHandlerContext ctx, RpcRequest req) { RpcResponse resp = new RpcResponse(); resp.setRequestId(req.getRequestId()); try { Object result = handle(req); resp.setResult(result); } catch (Throwable t) { resp.setError(t); } ctx.writeAndFlush(resp).addListener(ChannelFutureListener.CLOSE); } private Object handle(RpcRequest req) throws Throwable { Object bean = handlerMap.get(req.getClassName()); // CGLib FastClass 替代反射 FastClass fc = FastClass.create(bean.getClass()); FastMethod fm = fc.getMethod(req.getMethodName(), req.getParameterTypes()); return fm.invoke(bean, req.getParameters()); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { LOG.error("Handler error", cause); ctx.close(); } }

Consumer 端實作 (步驟 11-14 精簡版)

服務發現 (ServiceDiscovery - ZooKeeper Watcher)

public class ServiceDiscovery { private final String registryAddress; private volatile List<String> services = new ArrayList<>(); public ServiceDiscovery(String registryAddress) { this.registryAddress = registryAddress; ZooKeeper zk = connect(); if (zk != null) watch(zk); } public String discover() { if (services.isEmpty()) return null; return services.get(ThreadLocalRandom.current().nextInt(services.size())); // 簡易負載均衡 } // connect/watch 邏輯：獲取 /registry/data 子節點、註冊 Watcher 監聽變化、更新 services 列表 }

RPC 代理 (RpcProxy - JDK 動態代理)

public class RpcProxy { private final ServiceDiscovery discovery; public RpcProxy(ServiceDiscovery discovery) { this.discovery = discovery; } @SuppressWarnings("unchecked") public <T> T create(Class<?> interfaceClass) { return (T) Proxy.newProxyInstance( interfaceClass.getClassLoader(), new Class[]{interfaceClass}, (proxy, method, args) -> { RpcRequest req = new RpcRequest(); req.setRequestId(UUID.randomUUID().toString()); req.setClassName(method.getDeclaringClass().getName()); req.setMethodName(method.getName()); req.setParameterTypes(method.getParameterTypes()); req.setParameters(args); String serverAddr = discovery.discover(); if (serverAddr == null) throw new IllegalStateException("No available provider"); RpcClient client = new RpcClient(serverAddr); return client.send(req); }); } }

RPC 客戶端 (RpcClient - Netty 同步等待)

public class RpcClient extends SimpleChannelInboundHandler<RpcResponse> { private final String host; private final int port; private RpcResponse response; private final Object lock = new Object(); public RpcClient(String serverAddr) { String[] p = serverAddr.split(":"); this.host = p[0]; this.port = Integer.parseInt(p[1]); } public RpcResponse send(RpcRequest req) throws Exception { EventLoopGroup group = new NioEventLoopGroup(); try { Bootstrap b = new Bootstrap(); b.group(group).channel(NioSocketChannel.class) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) { ch.pipeline() .addLast(new RpcEncoder(RpcRequest.class)) .addLast(new RpcDecoder(RpcResponse.class)) .addLast(RpcClient.this); } }) .option(ChannelOption.SO_KEEPALIVE, true); ChannelFuture f = b.connect(host, port).sync(); f.channel().writeAndFlush(req).sync(); synchronized (lock) { lock.wait(); // 等待響應 } return response; } finally { group.shutdownGracefully(); } } @Override protected void channelRead0(ChannelHandlerContext ctx, RpcResponse msg) { this.response = msg; synchronized (lock) { lock.notifyAll(); } } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { LOG.error("Client error", cause); ctx.close(); } }

Consumer Spring 配置

<beans> <context:property-placeholder location="classpath:config.properties"/> <bean id="serviceDiscovery" class="com.example.rpc.registry.ServiceDiscovery"> <constructor-arg value="${registry.address}"/> </bean> <bean id="rpcProxy" class="com.example.rpc.client.RpcProxy"> <constructor-arg ref="serviceDiscovery"/> </bean> </beans>

單元測試驗證 (JUnit + Spring)

@RunWith(SpringJUnit4ClassRunner.class) @ContextConfiguration("classpath:spring.xml") public class HelloServiceTest { @Autowired private RpcProxy rpcProxy; @Test public void testHello() { HelloService service = rpcProxy.create(HelloService.class); String result = service.hello("World"); Assert.assertEquals("Hello! World", result); } }

完整 Maven 依賴 (pom.xml)

<dependencies> <!-- JUnit --> <dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>4.11</version><scope>test</scope></dependency> <!-- SLF4J + Log4j --> <dependency><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId><version>1.7.7</version></dependency> <!-- Spring --> <dependency><groupId>org.springframework</groupId><artifactId>spring-context</artifactId><version>3.2.12.RELEASE</version></dependency> <dependency><groupId>org.springframework</groupId><artifactId>spring-test</artifactId><version>3.2.12.RELEASE</version><scope>test</scope></dependency> <!-- Netty --> <dependency><groupId>io.netty</groupId><artifactId>netty-all</artifactId><version>4.0.24.Final</version></dependency> <!-- Protostuff --> <dependency><groupId>com.dyuproject.protostuff</groupId><artifactId>protostuff-core</artifactId><version>1.0.8</version></dependency> <dependency><groupId>com.dyuproject.protostuff</groupId><artifactId>protostuff-runtime</artifactId><version>1.0.8</version></dependency> <!-- ZooKeeper --> <dependency><groupId>org.apache.zookeeper</groupId><artifactId>zookeeper</artifactId><version>3.4.6</version></dependency> <!-- Commons Collections --> <dependency><groupId>org.apache.commons</groupId><artifactId>commons-collections4</artifactId><version>4.0</version></dependency> <!-- Objenesis --> <dependency><groupId>org.objenesis</groupId><artifactId>objenesis</artifactId><version>2.1</version></dependency> <!-- CGLib --> <dependency><groupId>cglib</groupId><artifactId>cglib</artifactId><version>3.1</version></dependency> </dependencies>

總結與生產級演進方向

本框架實現了 RPC 核心閉環：服務定義 → 提供者暴露 → 註冊中心註冊 → 消費者發現 → 動態代理 → Netty 異步通訊 → Protostuff 高效序列化 → CGLib 快速調用 → 響應返回。
生產級演進建議：
① 負載均衡：加權輪詢/最少連接/一致性哈希
② 容錯機制：重試、超時、熔斷 (Hystrix/Resilience4j)、降級
③ 流量治理：限流、路由規則、灰度發布
④ 序列化可插拔：SPI 機制支援 Hessian/Kryo/Protobuf/JSON
⑤ 多協議支援：HTTP/REST、gRPC、Dubbo 協議兼容
⑥ 監控指標：Micrometer + Prometheus + Grafana
⑦ 配置中心：Apollo/Nacos 取代 properties
願這份從零構建的經歷，成為你理解 Dubbo/gRPC 等成熟框架內核的最佳註腳。