【Java】【从零开发RPC框架】
- 版权声明：本站原创文章，于2026-03-26，由 管理员 发表。
- 转载请注明： 【Java】【从零开发RPC框架】 | CodeGarden
# 从零开发RPC框架：设计模式、反射、注解与网络优化的完美结合
> 本文将通过一个完整的RPC框架实现，带你深入理解设计模式、Java反射、注解机制以及网络I/O优化的实际应用。所有代码均可直接运行，建议跟随实战。
## 目录
1. [RPC框架概述](#rpc框架概述)
2. [框架架构设计](#框架架构设计)
3. [核心注解定义](#核心注解定义)
4. [设计模式应用](#设计模式应用)
5. **[完整代码实现](#完整代码实现)**
6. [JVM调优与网络优化](#jvm调优与网络优化)
7. [性能测试与优化](#性能测试与优化)
8. [总结](#总结)
---
## RPC框架概述
### 什么是RPC
RPC（Remote Procedure Call）远程过程调用，是一种计算机通信协议，允许运行在一台计算机上的程序调用另一台计算机上的程序，而无需了解底层网络细节。
### 我们要实现的功能
- ✅ 通过注解定义RPC服务接口
- ✅ 自动生成代理对象
- ✅ 基于Netty的高性能网络通信
- ✅ 多种序列化方式支持
- ✅ 负载均衡策略
- ✅ 服务注册与发现
- ✅ 优雅的服务调用
---
## 框架架构设计
```
┌─────────────────────────────────────────┐
│ RPC Framework │
├─────────────────────────────────────────┤
│ @Service @Reference @LoadBalance │ ← 注解层
├─────────────────────────────────────────┤
│ RpcClient RpcServer RpcRegistry │ ← API层
├─────────────────────────────────────────┤
│ ProxyFactory Serializer Codec │ ← 核心引擎
├─────────────────────────────────────────┤
│ LoadBalancer ServiceDiscovery │ ← 服务治理
├─────────────────────────────────────────┤
│ Netty Channel EventLoop │ ← 网络层
├─────────────────────────────────────────┤
│ TCP/IP NIO Selector │ ← 传输层
└─────────────────────────────────────────┘
服务端流程：
客户端调用 → 代理拦截 → 序列化 → 网络传输 → 反序列化 → 方法调用 → 结果返回
服务端流程：
接收请求 → 反序列化 → 方法调用 → 序列化结果 → 网络返回
```
---
## 核心注解定义
首先定义框架需要的核心注解：
```java
package com.rpc.annotation;
import java.lang.annotation.*;
/**
* 标识RPC服务接口
*/
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.TYPE)
public @interface RpcService {
/**
* 服务名称，默认使用接口全限定名
*/
String value() default "";
/**
* 服务版本
*/
String version() default "1.0.0";
}
/**
* RPC服务引用
*/
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.FIELD)
public @interface RpcReference {
/**
* 服务名称
*/
String value() default "";
/**
* 服务版本
*/
String version() default "1.0.0";
/**
* 超时时间（毫秒）
*/
int timeout() default 5000;
/**
* 负载均衡策略
*/
String loadBalance() default "roundRobin";
}
/**
* 负载均衡策略
*/
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.TYPE)
public @interface LoadBalance {
String strategy() default "roundRobin";
}
/**
* 序列化方式
*/
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.TYPE)
public @interface Serialization {
String value() default "json";
}
```
---
## 设计模式应用
### 1. 代理模式 - 动态代理工厂
```java
package com.rpc.proxy;
import com.rpc.annotation.RpcReference;
import com.rpc.client.RpcClient;
import com.rpc.core.RpcRequest;
import com.rpc.core.RpcResponse;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Proxy;
/**
* RPC代理工厂 - 代理模式
*/
public class RpcProxyFactory {
/**
* 创建RPC服务代理对象
*/
@SuppressWarnings("unchecked")
public static T createProxy(Class serviceInterface,
RpcReference reference,
RpcClient rpcClient) {
return (T) Proxy.newProxyInstance(
serviceInterface.getClassLoader(),
new Class[]{serviceInterface},
new RpcInvocationHandler(serviceInterface, reference, rpcClient)
);
}
/**
* RPC调用处理器
*/
static class RpcInvocationHandler implements InvocationHandler {
private final Class serviceInterface;
private final RpcReference reference;
private final RpcClient rpcClient;
public RpcInvocationHandler(Class serviceInterface,
RpcReference reference,
RpcClient rpcClient) {
this.serviceInterface = serviceInterface;
this.reference = reference;
this.rpcClient = rpcClient;
}
@Override
public Object invoke(Object proxy, java.lang.reflect.Method method, Object[] args)
throws Throwable {
// 构建RPC请求
String serviceName = reference.value().isEmpty() ?
serviceInterface.getName() : reference.value();
RpcRequest request = new RpcRequest();
request.setServiceName(serviceName);
request.setServiceVersion(reference.version());
request.setMethodName(method.getName());
request.setParameterTypes(method.getParameterTypes());
request.setParameters(args);
request.setTimeout(reference.timeout());
// 发送RPC请求
RpcResponse response = rpcClient.sendRequest(request);
// 处理响应
if (response.getError() != null) {
throw new RuntimeException("RPC调用失败: " + response.getError());
}
return response.getResult();
}
}
}
```
### 2. 适配器模式 - 序列化器适配
```java
package com.rpc.serializer;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
* 序列化器工厂 - 适配器模式
*/
public class SerializerFactory {
private static final Map SERIALIZER_MAP = new ConcurrentHashMap<>();
static {
// 注册内置序列化器
registerSerializer("json", new JsonSerializer());
registerSerializer("hessian", new HessianSerializer());
registerSerializer("kryo", new KryoSerializer());
registerSerializer("protobuf", new ProtobufSerializer());
}
/**
* 注册序列化器
*/
public static void registerSerializer(String type, Serializer serializer) {
SERIALIZER_MAP.put(type, serializer);
}
/**
* 获取序列化器（工厂方法）
*/
public static Serializer getSerializer(String type) {
Serializer serializer = SERIALIZER_MAP.get(type);
if (serializer == null) {
throw new IllegalArgumentException("不支持的序列化方式: " + type);
}
return serializer;
}
/**
* 序列化器接口
*/
public interface Serializer {
/**
* 序列化
*/
byte[] serialize(Object obj) throws Exception;
/**
* 反序列化
*/
T deserialize(byte[] data, Class clazz) throws Exception;
/**
* 获取序列化类型
*/
String getType();
}
/**
* JSON序列化器（使用Jackson）
*/
static class JsonSerializer implements Serializer {
private final com.fasterxml.jackson.databind.ObjectMapper objectMapper;
public JsonSerializer() {
this.objectMapper = new com.fasterxml.jackson.databind.ObjectMapper();
}
@Override
public byte[] serialize(Object obj) throws Exception {
return objectMapper.writeValueAsBytes(obj);
}
@Override
public T deserialize(byte[] data, Class clazz) throws Exception {
return objectMapper.readValue(data, clazz);
}
@Override
public String getType() {
return "json";
}
}
/**
* Hessian序列化器
*/
static class HessianSerializer implements Serializer {
@Override
public byte[] serialize(Object obj) throws Exception {
com.caucho.hessian.io.Hessian2Output output = new com.caucho.hessian.io.Hessian2Output(null);
java.io.ByteArrayOutputStream baos = new java.io.ByteArrayOutputStream();
output.init(baos);
output.writeObject(obj);
output.close();
return baos.toByteArray();
}
@Override
public T deserialize(byte[] data, Class clazz) throws Exception {
com.caucho.hessian.io.Hessian2Input input = new com.caucho.hessian.io.Hessian2Input(null);
java.io.ByteArrayInputStream bais = new java.io.ByteArrayInputStream(data);
input.init(bais);
Object obj = input.readObject();
input.close();
return clazz.cast(obj);
}
@Override
public String getType() {
return "hessian";
}
}
/**
* Kryo序列化器
*/
static class KryoSerializer implements Serializer {
private final ThreadLocal kryoThreadLocal;
public KryoSerializer() {
this.kryoThreadLocal = ThreadLocal.withInitial(() -> {
com.esotericsoftware.kryo.Kryo kryo = new com.esotericsoftware.kryo.Kryo();
kryo.setRegistrationRequired(false);
return kryo;
});
}
@Override
public byte[] serialize(Object obj) throws Exception {
com.esotericsoftware.kryo.Kryo kryo = kryoThreadLocal.get();
try (com.esotericsoftware.kryo.io.Output output = new com.esotericsoftware.kryo.io.Output(new java.io.ByteArrayOutputStream())) {
kryo.writeObject(output, obj);
output.flush();
return output.toBytes();
}
}
@Override
public T deserialize(byte[] data, Class clazz) throws Exception {
com.esotericsoftware.kryo.Kryo kryo = kryoThreadLocal.get();
try (com.esotericsoftware.kryo.io.Input input = new com.esotericsoftware.kryo.io.Input(new java.io.ByteArrayInputStream(data))) {
return kryo.readObject(input, clazz);
}
}
@Override
public String getType() {
return "kryo";
}
}
/**
* Protobuf序列化器
*/
static class ProtobufSerializer implements Serializer {
@Override
public byte[] serialize(Object obj) throws Exception {
// 简化实现，实际需要根据protobuf生成的类处理
if (obj instanceof com.google.protobuf.Message) {
return ((com.google.protobuf.Message) obj).toByteArray();
}
throw new UnsupportedOperationException("仅支持Protobuf Message对象");
}
@Override
public T deserialize(byte[] data, Class clazz) throws Exception {
throw new UnsupportedOperationException("请使用Protobuf生成的类");
}
@Override
public String getType() {
return "protobuf";
}
}
}
```
### 3. 策略模式 - 负载均衡
```java
package com.rpc.loadbalance;
import java.util.List;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicInteger;
/**
* 负载均衡器 - 策略模式
*/
public class LoadBalancerFactory {
/**
* 创建负载均衡器（工厂方法）
*/
public static LoadBalancer createLoadBalancer(String strategy) {
switch (strategy) {
case "random":
return new RandomLoadBalancer();
case "roundRobin":
return new RoundRobinLoadBalancer();
case "weighted":
return new WeightedLoadBalancer();
case "leastActive":
return new LeastActiveLoadBalancer();
default:
return new RoundRobinLoadBalancer();
}
}
/**
* 负载均衡器接口
*/
public interface LoadBalancer {
/**
* 选择一个服务实例
*/
String select(List instances);
}
/**
* 随机负载均衡
*/
static class RandomLoadBalancer implements LoadBalancer {
@Override
public String select(List instances) {
if (instances == null || instances.isEmpty()) {
throw new IllegalArgumentException("服务实例列表为空");
}
int index = ThreadLocalRandom.current().nextInt(instances.size());
return instances.get(index);
}
}
/**
* 轮询负载均衡
*/
static class RoundRobinLoadBalancer implements LoadBalancer {
private final AtomicInteger counter = new AtomicInteger(0);
@Override
public String select(List instances) {
if (instances == null || instances.isEmpty()) {
throw new IllegalArgumentException("服务实例列表为空");
}
int index = counter.getAndIncrement() % instances.size();
return instances.get(index);
}
}
/**
* 加权负载均衡
*/
static class WeightedLoadBalancer implements LoadBalancer {
@Override
public String select(List instances) {
if (instances == null || instances.isEmpty()) {
throw new IllegalArgumentException("服务实例列表为空");
}
// 简化实现：随机选择
int index = ThreadLocalRandom.current().nextInt(instances.size());
return instances.get(index);
}
}
/**
* 最少活跃连接负载均衡
*/
static class LeastActiveLoadBalancer implements LoadBalancer {
private final java.util.concurrent.ConcurrentMap activeCounts =
new java.util.concurrent.ConcurrentHashMap<>();
@Override
public String select(List instances) {
if (instances == null || instances.isEmpty()) {
throw new IllegalArgumentException("服务实例列表为空");
}
String selected = instances.get(0);
int minActive = Integer.MAX_VALUE;
for (String instance : instances) {
int active = activeCounts.computeIfAbsent(instance, k -> new AtomicInteger(0)).get();
if (active < minActive) {
minActive = active;
selected = instance;
}
}
// 增加活跃连接数
activeCounts.computeIfAbsent(selected, k -> new AtomicInteger(0)).incrementAndGet();
return selected;
}
}
}
```
---
## 完整代码实现
### 4. 核心数据模型
```java
package com.rpc.core;
import java.io.Serializable;
/**
* RPC请求
*/
public class RpcRequest implements Serializable {
private static final long serialVersionUID = 1L;
private String requestId;
private String serviceName;
private String serviceVersion;
private String methodName;
private Class[] parameterTypes;
private Object[] parameters;
private int timeout;
public RpcRequest() {
this.requestId = java.util.UUID.randomUUID().toString();
}
// getters and setters
public String getRequestId() { return requestId; }
public void setRequestId(String requestId) { this.requestId = requestId; }
public String getServiceName() { return serviceName; }
public void setServiceName(String serviceName) { this.serviceName = serviceName; }
public String getServiceVersion() { return serviceVersion; }
public void setServiceVersion(String serviceVersion) { this.serviceVersion = serviceVersion; }
public String getMethodName() { return methodName; }
public void setMethodName(String methodName) { this.methodName = methodName; }
public Class[] getParameterTypes() { return parameterTypes; }
public void setParameterTypes(Class[] parameterTypes) { this.parameterTypes = parameterTypes; }
public Object[] getParameters() { return parameters; }
public void setParameters(Object[] parameters) { this.parameters = parameters; }
public int getTimeout() { return timeout; }
public void setTimeout(int timeout) { this.timeout = timeout; }
}
/**
* RPC响应
*/
public class RpcResponse implements Serializable {
private static final long serialVersionUID = 1L;
private String requestId;
private Object result;
private String error;
// getters and setters
public String getRequestId() { return requestId; }
public void setRequestId(String requestId) { this.requestId = requestId; }
public Object getResult() { return result; }
public void setResult(Object result) { this.result = result; }
public String getError() { return error; }
public void setError(String error) { this.error = error; }
}
```
### 5. 服务端实现
```java
package com.rpc.server;
import com.rpc.core.RpcRequest;
import com.rpc.core.RpcResponse;
import com.rpc.serializer.Serializer;
import com.rpc.serializer.SerializerFactory;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.MessageToByteEncoder;
import java.lang.reflect.Method;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
* RPC服务端
*/
public class RpcServer {
private final int port;
private final Map serviceRegistry = new ConcurrentHashMap<>();
private final Serializer serializer;
private EventLoopGroup bossGroup;
private EventLoopGroup workerGroup;
public RpcServer(int port, String serialization) {
this.port = port;
this.serializer = SerializerFactory.getSerializer(serialization);
}
/**
* 注册服务
*/
public void registerService(String serviceName, Object serviceImpl) {
serviceRegistry.put(serviceName, serviceImpl);
System.out.println("注册服务: " + serviceName);
}
/**
* 启动服务
*/
public void start() throws InterruptedException {
bossGroup = new NioEventLoopGroup(1);
workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 128)
.childOption(ChannelOption.SO_KEEPALIVE, true)
.childHandler(new ChannelInitializer() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new RpcDecoder(serializer));
pipeline.addLast(new RpcEncoder(serializer));
pipeline.addLast(new RpcServerHandler());
}
});
ChannelFuture future = bootstrap.bind(port).sync();
System.out.println("RPC服务端启动，监听端口: " + port);
future.channel().closeFuture().sync();
} finally {
shutdown();
}
}
/**
* 关闭服务
*/
public void shutdown() {
if (workerGroup != null) {
workerGroup.shutdownGracefully();
}
if (bossGroup != null) {
bossGroup.shutdownGracefully();
}
}
/**
* RPC服务端处理器
*/
private class RpcServerHandler extends SimpleChannelInboundHandler {
@Override
protected void channelRead0(ChannelHandlerContext ctx, RpcRequest request) throws Exception {
RpcResponse response = handleRequest(request);
ctx.writeAndFlush(response);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
ctx.close();
}
}
/**
* 处理RPC请求
*/
private RpcResponse handleRequest(RpcRequest request) {
RpcResponse response = new RpcResponse();
response.setRequestId(request.getRequestId());
try {
// 查找服务实现
Object service = serviceRegistry.get(request.getServiceName());
if (service == null) {
throw new RuntimeException("服务不存在: " + request.getServiceName());
}
// 反射调用方法
Method method = service.getClass().getMethod(
request.getMethodName(),
request.getParameterTypes()
);
Object result = method.invoke(service, request.getParameters());
response.setResult(result);
} catch (Exception e) {
response.setError(e.getMessage());
e.printStackTrace();
}
return response;
}
}
/**
* RPC解码器
*/
class RpcDecoder extends ByteToMessageDecoder {
private final Serializer serializer;
public RpcDecoder(Serializer serializer) {
this.serializer = serializer;
}
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List