第05篇:手写JavaRPC框架之执行层思路
作者: 西魏陶渊明
博客: https://blog.springlearn.cn/
天下代码一大抄, 抄来抄去有提高, 看你会抄不会抄！
一、思路分析
通过前四篇文章的一起 Coding, 我们已经完成了 30%的工作,即完成了一个通信层的搭建。在不依赖任何 web 容器的情况下,可以实现服务之间的通信工作。就像下面这样。
客户端&服务端通信
1 /**
2 * @author liuxin
3 * 个人博客：https://java.springlearn.cn
4 * 公众号：西魏陶渊明 ｛关注获取学习源码｝
5 * 2022/8/11 23:12
6 */
7 @Test
8 @DisplayName("构建服务端【阻塞方式】")
9 public void server() throws Exception {
10 Mojito.server(RpcRequest.class, RpcResponse.class)
11 // 业务层,读取请求对象,返回结果
12 .businessHandler((channelContext, request) -> new RpcResponse())
13 .create().start(666);
14 }
15
16 @Test
17 @DisplayName("构建客户端【同步方式】")
18 public void clientSync() throws Exception{
19 Client<RpcRequest, RpcResponse> client = Mojito.client(RpcRequest.class, RpcResponse.class)
20 .connect("127.0.0.1", 6666);
21 System.out.println(client.send(new RpcRequest()));
22 }这只是完成了通信, 就好比,两台服务器之间建立了沟通管道,但是究竟怎么用这个管道呢? 如何将客户端的请求参数发送到服务端的服务器上执行结果,并返回给客户端呢? 这就是本篇文章要讨论的话题。
暂且我们把这一层叫做 RPC 执行层吧。按照老套路,在真正开始 Coding 之前,我们先梳理一下逻辑,画一个最基础的图。
- 左边客户端一个方法, 有 4 个服务端的实现。
- 右边某个服务端有一个具体的实现。
左边客户端的接口,通过代理的方式,将客户端的参数通过网络管道传输给某台服务端的本地进行执行。执行后获取结果,返回给客户端的调用方。而这些都是通过代理的方式实现的,所以开发者就好像调用本地方法一样。实现一次远程方法的调用。
以上是最最基础的远程调用的过程,但是如果就这的话就太基础了,下面我们会在这个基础上去做更多的事情。
1.1 客户端实现思路
RPC 服务,服务端会给客户端提供一个 API 包,这个包里面没有具体的实现,但是客户端能直接进行调用。学习 Java 的都知道接口是不能实例化的,但是为什么服务端给我们的 API,我们能直接调用呢? 当然是代理了。所以我们要学的第一个东西,就是学会代理。
- 使用代理实现接口的实例化。【代理是我们必须要掌握的】
但是代理层里面怎么做呢?
- 代理层将要调用的远程类和当前的客户端的参数,进行封装,然后通过通信层发送给服务端进行直接,然后拿到结果返回。这里面可能就设计到异步转同步的问题,但是没关系我们通信层直接就提供了实现。
以上就是客户端最基础的功能了。但是我们不满足于此, 市面上 RPC 框架有的功能,我们也要有。比如那些呢? 负载均衡、容错策略、事件广播我们也要有。那么他们究竟怎么做的呢?
-
负载均衡,我们看上面客户端的图,会发现这个服务有 4 个实现 [
172.168.10.1,172.168.10.2,172.168.10.3,172.168.10.4],说明服务端可能是集群部署的。那么既然有 4 个实例,我们就不能尽管这，一个 ip 进行进行调用,那么如何尽量让每个实例都能收到请求呢? 这就是负载均衡。 -
容错策略,当我们调用一台实例出错了,直接报错? 还是重试一下请求？或者是换一个实例在请求呢? 这就是容错策略。甚至我们还可以实现一个熔断器。
这里不会设计也没关系,我们会抄,哦,不对是借鉴。以下是dubbo的容错策略。
- 既然客户端有这个服务的所有实例信息,那么是不是不仅可以进行点对点的请求，还是能进行广播呢。 如下右图,同一个请求,如果发现是广播模式,就不在负载均衡的选其中的一个进行执行,而是每个都进行通知。
这五点就是客户端要具备的基础能力。除此之外客户端还要有断线重连, 自动剔除无效的链接的功能。极端情况也还要考虑当注册中心也挂了,无法找到服务端地址的场景,本地是不是也要维护一个服务连接的注册表。
1.2 服务端实现思路
服务端我们就可以参考 Web 容器的思路，比如 Servle 如何处理请求呢? http 过来就是一个 url,如何匹配到要执行的方法的呢?
- 服务启动将 API 实现保存到 Map 接口中。
- 服务收到请求,Dispatch 根据请求信息从 map 中获取执行器。
- 通过反射获取执行结果,返回给调用方。
同时将自己的信息注册到注册中心,让客户端能注册中心发现自己。
好了我们实现代码的大概思路就是这些,后面我们会一步一会去实现上面我们所说的功能。下面我们来看下 如果要实现上面的思路,还需要那些技术储备和软实力吧。
二、技术储备
要实现上面的想法,究竟需要哪些技术储备呢? 最基础的就是代理了。除此之外还要有架构设计能力。
2.1 代理模式
上面我们所说的负载均衡，容错策略，广播都要在代理中实现。
2.1.1 JDK 代理
JDK 代理必须要实现接口,第二个参数是接口数组,第三个参数是InvocationHandler。属于比较基础的内容。
1public class ProxyFactory {
2
3 public interface User {
4 String queryName();
5 }
6
7 public static void main(String[] args) {
8 User user = (User) Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class[]{User.class}, new InvocationHandler() {
9 @Override
10 public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
11 if (method.getName().equals("queryName")) {
12 return "Jay";
13 } else {
14 return null;
15 }
16 }
17 });
18 System.out.println(user.queryName());
19 }
20}2.1.2 Cglib 代理
Cglib 代理,代理类不用有接口。
1 public static class Person {
2
3 public String queryName() {
4 return "jay";
5 }
6 }
7
8 public static void main(String[] args) {
9 Person person = (Person) Enhancer.create(Person.class, new MethodInterceptor() {
10 @Override
11 public Object intercept(Object o, Method method, Object[] objects, MethodProxy methodProxy) throws Throwable {
12 if (method.getName().equals("queryName")) {
13 return "代理:Jay";
14 } else {
15 return null;
16 }
17 }
18 });
19 System.out.println(person.queryName());
20 }2.2 负载均衡策略
2.2.1 随机算法
1public static String random(List<String> services) {
2 Random random = new Random();
3 String[] addressArr = services.toArray(new String[0]);
4 // random
5 return addressArr[random.nextInt(services.size())];
6 }2.2.2 轮训算法
1public class RoundBalanceTest {
2
3 public static void main(String[] args) {
4 List<String> services = Arrays.asList("service1", "service2", "service3");
5 XxlBalanceTest.manyRoute(i -> {
6 // 请求次数,取模。serviceKey 可以更细粒度的控制轮训
7 ColorConsole.colorPrintln("轮训负载({}):{}", i, round(services));
8 });
9 }
10
11 private static final AtomicInteger atomicInteger = new AtomicInteger();
12
13 private static String round(List<String> services) {
14 int count = atomicInteger.get();
15 if (count >= Integer.MAX_VALUE) {
16 atomicInteger.set(0);
17 }
18 atomicInteger.incrementAndGet();
19 String[] toArray = services.toArray(new String[0]);
20 return toArray[count % toArray.length];
21 }
22}
232.2.3 加权算法
加权算法的有很多的变异算法, 可以通过配置的方式，也可以通过某种策略动态的给每台服务器进行加权，从而来提高被轮训到的次数。 这里说两种网上常见的实现。
简单加权算法 一个简单暴力的加权算法,如下图。按照权重，重新构建集合。然后再将集合进行取模轮训即可。即可实现一个最简单 的加权算法。
代码实现也是比较简单的,如下代码。
1public class WeightBalanceTest {
2
3 private static class Server {
4
5 private String host;
6
7 private Integer port;
8
9 public Server(String host, Integer port) {
10 this.host = host;
11 this.port = port;
12 }
13
14
15 }
16
17 private static final AtomicInteger atomicInteger = new AtomicInteger();
18
19 public static Server round(List<Server> services) {
20 int count = atomicInteger.get();
21 if (count >= Integer.MAX_VALUE) {
22 atomicInteger.set(0);
23 }
24 atomicInteger.incrementAndGet();
25 Server[] toArray = services.toArray(new Server[0]);
26 return toArray[count % toArray.length];
27 }
28
29 public static void main(String[] args) {
30 Map<Server, Integer> confWeight = new HashMap<>();
31 confWeight.put(new Server("127.0.0.1", 80), 2);
32 confWeight.put(new Server("127.0.0.1", 81), 3);
33 confWeight.put(new Server("127.0.0.1", 82), 5);
34
35 List<Server> servers = new ArrayList<>();
36 for (Map.Entry<Server, Integer> entity : confWeight.entrySet()) {
37 Server server = entity.getKey();
38 Integer weight = entity.getValue();
39 for (int i = 0; i < weight; i++) {
40 servers.add(server);
41 }
42 }
43 Loops.loop(10, i -> {
44 ColorConsole.colorPrintln("第{}次,权重轮训{}", i, round(servers));
45 });
46 }
47
48}
49第0次,权重轮训Server{host='127.0.0.1', port=80}
50第1次,权重轮训Server{host='127.0.0.1', port=80}
51第2次,权重轮训Server{host='127.0.0.1', port=82}
52第3次,权重轮训Server{host='127.0.0.1', port=82}
53第4次,权重轮训Server{host='127.0.0.1', port=82}
54第5次,权重轮训Server{host='127.0.0.1', port=82}
55第6次,权重轮训Server{host='127.0.0.1', port=82}
56第7次,权重轮训Server{host='127.0.0.1', port=81}
57第8次,权重轮训Server{host='127.0.0.1', port=81}
58第9次,权重轮训Server{host='127.0.0.1', port=81}
59但这样还是不均匀的, 相同的 ip 可能被连续的访问到其实就没有做到负载均衡。
平滑加权算法
主要解决上面那种不平滑的方案。这种方案是由 nginx (opens new window)提出来的。 算法的数学原理。
- 最大权重，减总权重
- 当前权重加上原权重 如下权重变化。
1| 轮数 | 选择前的当前权重 | 选择节点 | 选择后的当前权重 |
2| ---- | ---------------- | -------- | ---------------- |
3| 1 | {5, 1, 1} | a | {-2, 1, 1} |
4| 2 | {3, 2, 2} | a | {-4, 2, 2} |
5| 3 | {1, 3, 3} | b | {1, -4, 3} |
6| 4 | {6, -3, 4} | a | {-1, -3, 4} |
7| 5 | {4, -2, 5} | c | {4, -2, -2} |
8| 6 | {9, -1, -1} | a | {2, -1, -1} |
9| 7 | {7, 0, 0} | a | {0, 0, 0} |下面我们通过代码来实现。
- 首先我们定义出服务器模型,
weight是初始配置的权重，currentWeight是计算后的权重。 - 初始值
weight = currentWeight
1 @Data
2 @AllArgsConstructor
3 @ToString
4 @EqualsAndHashCode
5 private static class Server {
6 private String host;
7 private Integer port;
8 // 初始化权重
9 private Integer weight;
10 // 计算后的当前权重
11 private Integer currentWeight;
12 }然后我们根据算法的核心点来选择节点。这里我们先不考虑性能只说思路，有了思路在自己来优化代码。
- ==line(3-6)== 首先获取总权重
- ==line(8-14)== 然后获取当前最大权重的节点
- ==line(16-21)== 重新计算权重(
主要使用算法的思想)- 当前最大权重节点，重新计算权重。当前权重 = 当前权重 - 总权重 + 原始权重
- 其他节点，重新计算权重。当前权重 = 当前权重 + 原始权重
1 public static Server selectServer(List<Server> servers) {
2 // 获取总权重
3 Integer totalWeight = 0;
4 for (Server server : servers) {
5 totalWeight += server.getWeight();
6 }
7 // 根据权重从小到大排序
8 List<Server> sortByCurrentWeight = servers.stream().sorted(Comparator.comparing(Server::getCurrentWeight))
9 .collect(Collectors.toList());
10 // 集合反转,从大到小排序
11 Collections.reverse(sortByCurrentWeight);
12 // 当前最大权重
13 Server maxWeightServer = sortByCurrentWeight.get(0);
14 // 重新计算权重
15 for (Server server : servers) {
16 if (server.equals(maxWeightServer)) {
17 server.setCurrentWeight(server.getCurrentWeight() - totalWeight);
18 }
19 server.setCurrentWeight(server.getCurrentWeight() + server.getWeight());
20 }
21 return maxWeightServer;
22 }可以看到非常的平滑均匀，每个 ip 都会被分散。
1第0次,平滑权重轮训WeightBalanceTest2.Server(host=127,0,0,1, port=8080, weight=4, currentWeight=1)
2第1次,平滑权重轮训WeightBalanceTest2.Server(host=127,0,0,1, port=8081, weight=2, currentWeight=-1)
3第2次,平滑权重轮训WeightBalanceTest2.Server(host=127,0,0,1, port=8080, weight=4, currentWeight=2)
4第3次,平滑权重轮训WeightBalanceTest2.Server(host=127,0,0,1, port=8082, weight=1, currentWeight=-2)
5第4次,平滑权重轮训WeightBalanceTest2.Server(host=127,0,0,1, port=8080, weight=4, currentWeight=3)
6第5次,平滑权重轮训WeightBalanceTest2.Server(host=127,0,0,1, port=8081, weight=2, currentWeight=0)
7第6次,平滑权重轮训WeightBalanceTest2.Server(host=127,0,0,1, port=8080, weight=4, currentWeight=4)很难想象,手写一个 RPC 框架,执行层所需要的技术储备只有这些而已。可以看出来，重要的是思路，思路决定出路。好了,下面我们就来实现这些思路吧。
那么你准备好跟我一起 Coding 了吗?
