网站title标签内容怎么设置小程序如何推广运营
我的技术网站 java-broke.site，有大厂完整面经，工作技术，架构师成长之路，等经验分享
Linkis-RPC的设计目标是提供一种灵活、可扩展的微服务间通信机制，支持以下功能：
- 异步请求与响应：支持请求方发送异步请求，接收方处理完后再响应给请求方。
- 广播请求：支持将请求广播给所有目标服务实例。
- 自定义拦截器：允许用户定义拦截器来实现特定功能，如缓存、重试等。
- 服务选择：基于Eureka实现的服务发现，支持通过服务名和实例信息选择特定服务实例。
- 解耦通信机制：请求方和接收方的代码实现解耦，便于独立开发和维护。
Linkis-RPC的架构概述
Linkis-RPC主要由以下几个核心组件构成：
- Sender：请求发送器，用于发送请求到目标服务。
- Receiver：请求接收器，负责接收和处理请求。
- Interceptor：拦截器链，用于在请求发送前进行处理，如广播、重试、缓存等。
- Decoder/Encoder：用于请求和响应的序列化和反序列化。
- Eureka：服务注册与发现中心。
Linkis-RPC的代码结构
Linkis-RPC的源码结构清晰，主要代码模块包括：
- Sender接口和实现：负责发送请求并处理响应。
- Receiver接口和实现：负责接收请求并执行业务逻辑。
- Interceptor接口和实现：拦截器实现类，用于增强请求功能。
- RPCReceiveRestful：内嵌的HTTP服务，用于接收和解码请求。
- 请求和响应编码器/解码器：实现请求和响应的序列化和反序列化。
接下来，我们将逐步分析每个模块的代码实现。
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