Linkis-RPC的设计思想

我的技术网站 java-broke.site，有大厂完整面经，工作技术，架构师成长之路，等经验分享

Linkis-RPC的设计目标是提供一种灵活、可扩展的微服务间通信机制，支持以下功能：

1. 异步请求与响应：支持请求方发送异步请求，接收方处理完后再响应给请求方。
2. 广播请求：支持将请求广播给所有目标服务实例。
3. 自定义拦截器：允许用户定义拦截器来实现特定功能，如缓存、重试等。
4. 服务选择：基于Eureka实现的服务发现，支持通过服务名和实例信息选择特定服务实例。
5. 解耦通信机制：请求方和接收方的代码实现解耦，便于独立开发和维护。

Linkis-RPC的架构概述

Linkis-RPC主要由以下几个核心组件构成：

• Sender：请求发送器，用于发送请求到目标服务。
• Receiver：请求接收器，负责接收和处理请求。
• Interceptor：拦截器链，用于在请求发送前进行处理，如广播、重试、缓存等。
• Decoder/Encoder：用于请求和响应的序列化和反序列化。
• Eureka：服务注册与发现中心。

Linkis-RPC的代码结构

Linkis-RPC的源码结构清晰，主要代码模块包括：

1. Sender接口和实现：负责发送请求并处理响应。
2. Receiver接口和实现：负责接收请求并执行业务逻辑。
3. Interceptor接口和实现：拦截器实现类，用于增强请求功能。
4. RPCReceiveRestful：内嵌的HTTP服务，用于接收和解码请求。
5. 请求和响应编码器/解码器：实现请求和响应的序列化和反序列化。

接下来，我们将逐步分析每个模块的代码实现。

Sender的实现

Sender接口

Sender接口提供了多种发送请求的方法，包括同步和异步请求。

public abstract class Sender {/*** 同步请求方法，等待接收方返回响应。** @param message 请求消息对象* @return 响应对象*/public abstract Object ask(Object message);/*** 带超时设置的同步请求方法。** @param message 请求消息对象* @param timeout 超时时间* @return 响应对象*/public abstract Object ask(Object message, Duration timeout);/*** 仅发送请求，不关心响应。** @param message 请求消息对象*/public abstract void send(Object message);/*** 异步请求方法，在稍后通过其他线程发送请求。** @param message 请求消息对象*/public abstract void deliver(Object message);
}

Sender的实现类

Sender的实现类主要负责构建请求并通过Feign客户端发送请求。

public class DefaultSender extends Sender {private final FeignClient feignClient;private final String serviceName;public DefaultSender(String serviceName, FeignClient feignClient) {this.serviceName = serviceName;this.feignClient = feignClient;}@Overridepublic Object ask(Object message) {return feignClient.post(message);}@Overridepublic Object ask(Object message, Duration timeout) {// 设置请求超时逻辑return feignClient.postWithTimeout(message, timeout);}@Overridepublic void send(Object message) {feignClient.send(message);}@Overridepublic void deliver(Object message) {// 异步发送逻辑CompletableFuture.runAsync(() -> feignClient.post(message));}
}

Receiver的实现

Receiver接口

Receiver接口定义了接收请求和响应的基本方法。

public interface Receiver {/*** 处理异步请求的方法。** @param message 请求消息对象* @param sender  请求发送方的Sender实例*/void receive(Object message, Sender sender);/*** 处理同步请求的方法，返回响应对象。** @param message 请求消息对象* @param sender  请求发送方的Sender实例* @return 响应对象*/Object receiveAndReply(Object message, Sender sender);/*** 带超时设置的同步请求处理方法。** @param message 请求消息对象* @param duration 超时时间* @param sender  请求发送方的Sender实例* @return 响应对象*/Object receiveAndReply(Object message, Duration duration, Sender sender);
}

Receiver的实现类

public class DefaultReceiver implements Receiver {@Overridepublic void receive(Object message, Sender sender) {// 异步处理逻辑System.out.println("Received async message: " + message);// 根据业务需求决定是否需要响应发送方}@Overridepublic Object receiveAndReply(Object message, Sender sender) {// 处理请求并返回响应System.out.println("Received sync message: " + message);return processRequest(message);}@Overridepublic Object receiveAndReply(Object message, Duration duration, Sender sender) {// 处理请求并返回响应，考虑超时System.out.println("Received sync message with timeout: " + message);return processRequestWithTimeout(message, duration);}private Object processRequest(Object message) {// 业务逻辑处理return "Processed: " + message;}private Object processRequestWithTimeout(Object message, Duration duration) {// 业务逻辑处理，支持超时try {Thread.sleep(duration.toMillis());} catch (InterruptedException e) {Thread.currentThread().interrupt();return "Processing interrupted";}return "Processed with timeout: " + message;}
}

Interceptor的实现

拦截器接口

Interceptor接口定义了在请求发送前后的处理逻辑。

public interface RPCInterceptor {/*** 请求发送前的处理逻辑。** @param message 请求消息对象*/void preHandle(Object message);/*** 请求发送后的处理逻辑。** @param message 请求消息对象* @param response 响应对象*/void postHandle(Object message, Object response);
}

广播拦截器实现

广播拦截器用于将请求广播给所有服务实例。

@Componentpublic class BroadcastRPCInterceptor implements RPCInterceptor {@Overridepublic void preHandle(Object message) {if (message instanceof BroadcastMessage) {// 广播逻辑System.out.println("Broadcasting message: " + message);}}@Overridepublic void postHandle(Object message, Object response) {// 后处理逻辑}
}

重试拦截器实现

重试拦截器用于在请求失败时进行重试。

@Componentpublic class RetryableRPCInterceptor implements RPCInterceptor {@Overridepublic void preHandle(Object message) {// 重试逻辑处理}@Overridepublic void postHandle(Object message, Object response) {if (response instanceof Throwable) {System.out.println("Request failed, retrying...");// 重试逻辑}}
}

缓存拦截器实现

缓存拦截器用于对不频繁变动的响应进行缓存。

@Componentpublic class CacheableRPCInterceptor implements RPCInterceptor {private final Map<Object, Object> cache = new ConcurrentHashMap<>();@Overridepublic void preHandle(Object message) {if (message instanceof CacheableMessage) {// 检查缓存Object cachedResponse = cache.get(message);if (cachedResponse != null) {System.out.println("Cache hit: " + message);// 返回缓存响应}}}@Overridepublic void postHandle(Object message, Object response) {if (message instanceof CacheableMessage) {// 缓存响应cache.put(message, response);}}
}

自定义拦截器实现

用户可以根据需要实现自定义拦截器，实现特定功能。

@Componentpublic class CustomRPCInterceptor implements RPCInterceptor {@Overridepublic void preHandle(Object message) {// 自定义处理逻辑}@Overridepublic void postHandle(Object message, Object response) {// 自定义处理逻辑}}

请求和响应编码器/解码器

请求编码器

将请求对象序列化为JSON字符串。

public class RPCEncoder {public String encode(Object message) {// 使用Jackson或Gson进行序列化return new ObjectMapper().writeValueAsString(message);}
}

请求解码器

将JSON字符串反序列化为请求对象。

public class RPCDecoder {public <T> T decode(String json, Class<T> clazz) {// 使用Jackson或Gson进行反序列化return new ObjectMapper().readValue(json, clazz);}
}

RPCReceiveRestful的实现

RPCReceiveRestful是一个内嵌的HTTP服务，用于接收请求并调用解码器进行解码。

@RestController@RequestMapping("/rpc")
public class RPCReceiveRestful {private final RPCDecoder decoder;private final ReceiverManager receiverManager;public RPCReceiveRestful(RPCDecoder decoder, ReceiverManager receiverManager) {this.decoder = decoder;this.receiverManager = receiverManager;}@PostMapping("/receive")public ResponseEntity<Object> receiveRequest(@RequestBody String requestJson) {try {// 解码请求RPCRequest request = decoder.decode(requestJson, RPCRequest.class);// 获取对应的ReceiverReceiver receiver = receiverManager.getReceiver(request.getServiceName());// 调用Receiver处理请求Object response = receiver.receiveAndReply(request.getMessage(), request.getSender());return ResponseEntity.ok(response);} catch (Exception e) {// 请求解码或处理失败return ResponseEntity.status(HttpStatus.BAD_REQUEST).body("Request processing failed");}}
}

示例代码：使用Linkis-RPC进行通信

创建Sender

public class RpcClient {private final Sender sender;public RpcClient(Sender sender) {this.sender = sender;}public void sendMessage(String message) {// 异步发送消息sender.deliver(message);}public Object requestResponse(String message) {// 同步请求响应return sender.ask(message);}
}

创建Receiver

public class RpcServer implements Receiver {@Overridepublic void receive(Object message, Sender sender) {// 处理异步请求System.out.println("Server received async message: " + message);}@Overridepublic Object receiveAndReply(Object message, Sender sender) {// 处理同步请求并返回响应System.out.println("Server received sync message: " + message);return "Server response to: " + message;}@Overridepublic Object receiveAndReply(Object message, Duration duration, Sender sender) {// 处理带超时的同步请求并返回响应System.out.println("Server received sync message with timeout: " + message);return "Server response with timeout to: " + message;}
}

结论

Linkis-RPC提供了一种灵活且强大的微服务间通信机制，解决了传统RPC框架在复杂场景中的不足。通过自定义拦截器、异步请求处理和广播机制，Linkis-RPC能够满足现代微服务架构的通信需求。本文详细分析了Linkis-RPC的设计思想、代码结构，并提供了完整的代码示例，希望能为开发者提供有价值的参考。

如需进一步的讨论和问题解答，欢迎留言，共同探讨Linkis-RPC的更多应用场景和实现细节。

参考链接：公共模块 - RPC 模块 - 《Apache Linkis v1.3.0 中文文档》 - 书栈网 · BookStack