当前位置：首页 > ds >正文

用Java实现rpc的逻辑和流程图和核心技术与难点分析

ds 2025/7/27 12:03:22

以下为本人对RPC理解后的想法，不一定正确

1.流程图

2. 核心实现代码结构

2.1 接口定义

flowchart TDA[客户端调用] --> B[动态代理生成代理对象]B --> C[代理对象拦截方法调用]C --> D[序列化请求参数]D --> E[网络传输层]E --> F[服务端接收请求]F --> G[反序列化请求]G --> H[反射调用目标方法]H --> I[执行业务逻辑]I --> J[序列化响应结果]J --> K[网络返回响应]K --> L[客户端接收响应]L --> M[反序列化响应]M --> N[返回调用结果]E --> E1[连接池管理]E --> E2[负载均衡]E --> E3[服务发现]E --> E4[超时重试]H --> H1[参数校验]H --> H2[权限验证]H --> H3[异常处理]

2.2 服务端实现


public interface UserService {User getUserById(Long id);List<User> getUsers();}

2.3 RPC框架核心组件

2.3.1 服务注册与发现

@Service
public class UserServiceImpl implements UserService {@Overridepublic User getUserById(Long id) {// 业务逻辑return userRepository.findById(id);}
}

2.3.2网络传输层

@Component
public class ServiceRegistry {private Map<String, Object> serviceMap = new ConcurrentHashMap<>();public void register(String serviceName, Object service) {serviceMap.put(serviceName, service);}public Object getService(String serviceName) {return serviceMap.get(serviceName);}
}

2.3.3序列化和反序列化

public class NettyServer {private EventLoopGroup bossGroup;private EventLoopGroup workerGroup;public void start(int port) {ServerBootstrap bootstrap = new ServerBootstrap();bootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childHandler(new RpcServerInitializer());bootstrap.bind(port).sync();}
}

2.3.4动态代理

public class SerializationUtil {public static byte[] serialize(Object obj) {try (ByteArrayOutputStream baos = new ByteArrayOutputStream();ObjectOutputStream oos = new ObjectOutputStream(baos)) {oos.writeObject(obj);return baos.toByteArray();} catch (IOException e) {throw new RuntimeException("序列化失败", e);}}public static Object deserialize(byte[] bytes) {try (ByteArrayInputStream bais = new ByteArrayInputStream(bytes);ObjectInputStream ois = new ObjectInputStream(bais)) {return ois.readObject();} catch (Exception e) {throw new RuntimeException("反序列化失败", e);}}
}

3. 核心技术分析

3.1 网络通信

Netty框架：高性能异步网络框架

NIO：非阻塞IO，支持高并发

连接池：复用连接，减少连接开销

3.2 序列化技术

Java原生序列化：简单但性能较差

JSON：可读性好，跨语言

Protocol Buffers：高性能，跨语言

Hessian：轻量级，性能好

3.3 服务治理

服务注册与发现：ZooKeeper、Consul、Eureka

负载均衡：轮询、权重、一致性哈希

熔断降级：Hystrix、Sentinel

监控追踪：链路追踪、性能监控

4. 技术难点分析

4.1 网络通信难点

public class RpcProxy {public static <T> T create(Class<T> serviceClass) {return (T) Proxy.newProxyInstance(serviceClass.getClassLoader(),new Class<?>[]{serviceClass},new RpcInvocationHandler());}}class RpcInvocationHandler implements InvocationHandler {@Overridepublic Object invoke(Object proxy, Method method, Object[] args) {// 1. 构建RPC请求RpcRequest request = new RpcRequest();request.setClassName(method.getDeclaringClass().getName());request.setMethodName(method.getName());request.setParameters(args);request.setParameterTypes(method.getParameterTypes());// 2. 发送请求并获取响应RpcResponse response = sendRequest(request);// 3. 返回结果return response.getResult();}}

4.2 序列化难点

// 难点1：连接管理public class ConnectionManager {private Map<String, Channel> channelMap = new ConcurrentHashMap<>();public Channel getChannel(String address) {Channel channel = channelMap.get(address);if (channel == null || !channel.isActive()) {channel = createChannel(address);channelMap.put(address, channel);}return channel;}}// 难点2：异步处理public class AsyncRpcClient {private Map<String, CompletableFuture<RpcResponse>> pendingRequests = new ConcurrentHashMap<>();public CompletableFuture<RpcResponse> sendAsync(RpcRequest request) {CompletableFuture<RpcResponse> future = new CompletableFuture<>();pendingRequests.put(request.getRequestId(), future);// 发送请求return future;}}

4.3 异常处理难点

// 难点：处理复杂对象和循环引用public class AdvancedSerializer {public byte[] serialize(Object obj) {// 处理循环引用Map<Object, Integer> objectMap = new IdentityHashMap<>();return serializeInternal(obj, objectMap);}private byte[] serializeInternal(Object obj, Map<Object, Integer> objectMap) {if (objectMap.containsKey(obj)) {// 处理循环引用return writeReference(objectMap.get(obj));}// 正常序列化objectMap.put(obj, objectMap.size());return writeObject(obj);}}

4.4 性能优化难点

// 难点：网络异常、超时、服务异常的统一处理public class RpcExceptionHandler {public Object handleException(Exception e, RpcRequest request) {if (e instanceof TimeoutException) {// 超时重试return retry(request);} else if (e instanceof NetworkException) {// 网络异常，切换服务节点return switchNodeAndRetry(request);} else {// 业务异常，直接抛出throw new RpcException("RPC调用失败", e);}}}