偶然间看到一个RPC的框架项目，感觉与其他的项目与众不同，所以准备跟着实现一下。由于教程比较少，因此写下这边记录，慢慢踩坑。既然是重复造轮子，自然是以学习框架与组件的运用为主。
1 涉及的知识点
- RPC的概念及运作流程。
- RPC协议及RPC框架的概念。
- Netty的基本使用，修复高并发情况下，netty导致的内存泄漏问题。
- Java序列化以及反序列化，protobuf和kryo序列化协议，配置即用。
- Zookeeper作为注册中心的基本使用。
- 自定义注释实现特殊业务逻辑
- Java动态代理
- 自定义Spring Boot Starter
- 多种负载均衡算法（随机、轮询、加权轮询、平滑加权轮询），配置即用。
- 客户端增加本地服务列表缓存，提高性能。
- 由原来的每个请求建立一次连接，改为建立TCP长连接，并多次复用。
- 服务端增加线程池提高消息处理能力
2 RPC 基础知识
2.1 RPC是什么？
Remote Procedure Call（RPC）：远程过程调用。
RPC采用Client-Server结构，通过Request-Response消息模式实现。
2.2 RPC的流程
- 客户端处理过程中调用Client stub（就像调用本地方法一样），传递参数；
- Client stub将参数编组为消息，然后通过系统调用向服务端发送消息；
- 客户端本地操作系统将消息从客户端机器发送到服务端机器；
- 服务端操作系统将接收到的数据包传递给Server stub;
- Server stub解组消息为参数；
- Server stub再调用服务端的过程，过程执行结果以反方向的相同步骤响应给客户端。
2.3 RPC流程中需要处理的问题
- Client stub、Server stub的开发；
- 参数如何编组为消息，以及解组消息；
- 消息如何发送；
- 过程结果如何表示、异常情况如何处理；
- 如何实现安全的访问控制。
2.4 RPC协议是什么？
RPC调用过程中采用的消息协议称为RPC协议
RPC协议规定请求、响应消息的格式在TCP（网络传输控制协议）上可选用或自定义消息协议来完成RPC消息交互。我们可以选用通用的标准协议（如：http、https），也可以根据自身的需要定义自己的消息协议。
2.5 RPC框架是什么？
封装好参数编组、消息解组、底层网络通信的RPC程序开发框架，带来的便捷是可以直接在其基础上只需要专注于过程代码编写。
Java领域：
- 传统的webservice框架：Apache CXF、Apache Axis2、Java自带的JAX-WS等。webservice框架大多基于标准的SOAP协议。
- 新兴的微服务框架：Dubbo、spring cloud、Apache Thrift等。
3 手写RPC
3.1 目标
我们将会写一个简易的RPC框架，暂且叫它zarlic-rpc-spring-boot-starter，通过在项目中引入该starter，并简单的配置一下，项目即拥有提供远程服务的能力。
编写自定义注解@Service，被它注解的类将会提供远程服务。
编写自定义注解@InjectService，使用它可注入远程服务。
3.2 项目整体结构
- 注册中心 ：注册中心负责服务地址的注册与查找，相当于目录服务。
- 网络传输 ：既然我们要调用远程的方法，就要发送网络请求来传递目标类和方法的信息以及方法的参数等数据到服务提供端。
- 序列化和反序列化 ：要在网络传输数据就要涉及到序列化。
- 动态代理 ：屏蔽程方法调用的底层细节。
- 负载均衡 ： 避免单个服务器响应同一请求，容易造成服务器宕机、崩溃等问题。
- 传输协议 ：这个协议是客户端（服务消费方）和服务端（服务提供方）交流的基础。
3.3 客户端编写
3.3.1 客户端需要做什么
客户端想要调用远程服务，必须具备服务发现的能力；在知道有哪些服务过后，还必须有服务代理来执行服务调用；客户端想要与服务端通信，必须要有相同的消息协议；客户端想要调用远程服务，那么必须具备网络请求的能力，即网络层功能。
当然，这是客户端所需的最基本的能力，其实还可以扩展的能力，例如负载均衡。
3.3.2 具体实现
基于面向接口编程的理念，不同角色都实现了定义了相应规范的接口。这里面我们没有发现消息协议相关内容，那是因为服务端也需要消息协议，因此抽离了出来，放在公共层。
客户端的代码结构：
3.3.2.1 服务发现者
复制代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45/** * 服务发现抽象类，定义服务发现规范 */ public interface ServiceDiscoverer { List<Service> getService(String name); } /** Zookeeper服务发现者，定义以Zookeeper为注册中心的服务发现细则 * @author zarlic * @date 2021.12.15 15:28 */ public class ZookeeperServiceDiscoverer implements ServiceDiscoverer{ private ZkClient zkClient; public ZookeeperServiceDiscoverer(String zkAddress) { zkClient = new ZkClient(zkAddress); //配置zk中心地址 zkClient.setZkSerializer(new ZookeeperSerializer()); //自定义序列化，在common中定义ZookeeperSerializer } /** * 使用Zookeeper客户端，通过服务名获取服务列表 * 服务名格式：接口全路径 * * @param name 服务名 * @return 服务列表 */ @Override public List<Service> getService(String name) { String servicePath = ZarlicConstant.ZK_SERVICE_PATH + ZarlicConstant.PATH_DELIMITER + name + "/service"; List<String> children = zkClient.getChildren(servicePath); //ofNullable 如果children不为空就将其赋值ArrayList，为空创建一个空对象集合赋值给newList，也就避免了空指针异常。 //使用decode解码出服务名字，再根据服务名字，转换成对应的服务 return Optional.ofNullable(children).orElse(new ArrayList<>()).stream().map(str -> { String deCh = null; try { deCh = URLDecoder.decode(str,ZarlicConstant.UTF_8); } catch (UnsupportedEncodingException e) { e.printStackTrace(); } return JSON.parseObject(deCh,Service.class); }).collect(Collectors.toList()); } }
服务发现者使用Zookeeper来实现，通过ZkClient我们很容易发现已经注册在ZK上的服务。当然我们也可以使用其他组件作为注册中心，例如Redis。
3.3.2.2 网络客户端
复制代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129/**网络请求客户端，定义网络请求规范 * @author zarlic * @date 2021.12.15 15:29 */ public interface NetClient { /** * 客户端发送请求，即为Request，传送为序列化之后的数据流byte[] * @param data * @param service * @return * @throws InterruptedException */ byte[] sendRequest(byte[] data, Service service) throws InterruptedException; } /**Netty网络请求客户端，定义通过Netty实现网络请求的细则。 * @author zarlic * @date 2021.12.15 15:29 */ public class NettyNetClient implements NetClient{ /** * 日志调用 */ private static Logger logger = LoggerFactory.getLogger(NettyNetClient.class); @Override public byte[] sendRequest(byte[] data, Service service) throws InterruptedException { String[] addInfoArray = service.getAddress().split(":"); String serverAddress = addInfoArray[0]; String serverPort = addInfoArray[1]; SendHandler sendHandler = new SendHandler(data); byte[] respData; // 配置客户端 EventLoopGroup group = new NioEventLoopGroup(); try { Bootstrap bootstrap = new Bootstrap(); //group设置线程池，channel设置nio类型的channel，option设置通道选项，handler装配流水线 bootstrap.group(group).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY,true) .handler(new ChannelInitializer<SocketChannel>() { //有连接到达时会创建一个channel @Override protected void initChannel(SocketChannel socketChannel) throws Exception { ChannelPipeline channelPipeline = socketChannel.pipeline(); // pipeline管理channel中的Handler // 在channel队列中添加一个handler来处理业务 channelPipeline.addLast(sendHandler); } }); // 启动客户端连接 // 开始绑定server // 通过调用sync同步方法阻塞直到绑定成功 bootstrap.connect(serverAddress,Integer.parseInt(serverPort)).sync(); respData = (byte[]) sendHandler.rspData(); logger.info("SendRequest get reply:{}",respData); } finally { // 优雅关闭EventLoopGroup， // 释放掉所有资源包括创建的线程 group.shutdownGracefully(); } return respData; } } /** 发送处理类，定义Netty入站处理细则 * @author zarlic * @date 2021.12.15 15:28 */ public class SendHandler extends ChannelInboundHandlerAdapter { private static Logger logger = LoggerFactory.getLogger(SendHandler.class); private CountDownLatch countDownLatch; private Object readMsg = null; private byte[] data; public SendHandler(byte[] data) { countDownLatch = new CountDownLatch(1); this.data = data; } /** * 当连接服务端成功后，发送请求数据 * * @param ctx 通道上下文 */ @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { logger.info("Successful connection to server:{}",ctx); ByteBuf reqBuf = Unpooled.buffer(data.length); reqBuf.writeBytes(data); logger.info("Client send message:{}",reqBuf); ctx.writeAndFlush(reqBuf); } /** * 读取数据，数据读取完毕释放CD锁 * * @param ctx 上下文 * @param msg ByteBuf */ @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { logger.info("Client reads message: {}", msg); ByteBuf msgBuf = (ByteBuf) msg; byte[] resp = new byte[msgBuf.readableBytes()]; msgBuf.readBytes(resp); readMsg = resp; countDownLatch.countDown(); } @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { ctx.flush(); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { // Close the connection when an exception is raised. cause.printStackTrace(); logger.error("Exception occurred：{}", cause.getMessage()); ctx.close(); } /** * 等待读取数据完成 * * @return 响应数据 * @throws InterruptedException 异常 */ public Object rspData() throws InterruptedException { countDownLatch.await(); return readMsg; } }
在这里我们使用Netty来实现网络请求客户端，当然也可以使用Mina。网络请求客户端能连接远程服务端，并将编组好的请求数据发送给服务端，待服务端处理好后，又将服务端的响应数据返回给客户端。
3.3.2.3 服务代理
computeIfAbsent
复制代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15// 方法定义 default V computeIfAbsent(K key, Function<? super K, ? extends V> mappingFunction) { ... } // java8之前。从map中根据key获取value操作可能会有下面的操作 Object key = map.get("key"); if (key == null) { key = new Object(); map.put("key", key); } // java8之后。上面的操作可以简化为一行，若key对应的value为空，会将第二个参数的返回值存入并返回 Object key2 = map.computeIfAbsent("key", k -> new Object());
动态代理（dynamic proxy）
利用Java的反射技术(Java Reflection)，在运行时创建一个实现某些给定接口的新类（也称“动态代理类”）及其实例（对象）,代理的是接口(Interfaces)，不是类(Class)，也不是抽象类。在运行时才知道具体的实现，spring aop就是此原理。
复制代码
1
2
3
4
5public static Object newProxyInstance(ClassLoader loader, Class<?>[] interfaces, InvocationHandler h) throws IllegalArgumentException
newProxyInstance，方法有三个参数：
loader: 用哪个类加载器去加载代理对象
interfaces:动态代理类需要实现的接口
h:动态代理方法在执行时，会调用h里面的invoke方法去执行
复制代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87/** * 客户端代理工厂：用于创建远程服务代理类 * 封装编组请求、请求发送、编组响应等操作。 * @author zarlic * @date 2021.12.15 15:30 */ public class ClientProxyFactory { private ServiceDiscoverer serviceDiscoverer; private Map<String, MessageProtocol> supportMessageProtocols; private NetClient netClient; /** * 通过Java动态代理获取服务代理类 * * @param clazz 被代理类Class * @param <T> 泛型 * @return 服务代理类 */ @SuppressWarnings("unchecked") //该批注的作用是给编译器一条指令，告诉它对被批注的代码元素内部的某些警告保持静默。 private Map<Class<?>, Object> objectCache = new HashMap<>(); //获取代理类，若没有则使用java动态代理Proxy.newProxyInstance public <T> T getProxy(Class<T> clazz){ return (T) this.objectCache.computeIfAbsent(clazz, cls -> newProxyInstance(cls.getClassLoader(),new Class<?>[]{cls},new ClientInvocationHandler(cls))); } /** * 客户端服务代理类invoke函数细节实现 */ private class ClientInvocationHandler implements InvocationHandler{ private Class<?> clazz; private Random random = new Random(); public ClientInvocationHandler(Class<?> clazz){ super(); this.clazz = clazz; } @Override public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { if (method.getName().equals("toString")) { return proxy.getClass().toString(); } if (method.getName().equals("hashCode")) { return 0; } //1、获得服务信息 String serviceName = this.clazz.getName(); List<Service> services = serviceDiscoverer.getService(serviceName); if(services == null || services.isEmpty()){ throw new ZarlicException("No provider available!"); } //随机选择一个服务提供者 Service service = services.get(random.nextInt(services.size())); //2、构造request对象 ZarlicRequest req = new ZarlicRequest(); req.setServiceName(service.getName()); req.setMethod(method.getName()); req.setParameterTypes(method.getParameterTypes()); req.setParameters(args); //3、协议层编组 //获得该方法对应的协议 MessageProtocol protocol = supportMessageProtocols.get(service.getProtocol()); //编组请求 byte[] data = protocol.marshallingRequest(req); //4、调用网络层发送请求 byte[] repData = netClient.sendRequest(data, service); //5、解组相应消息 ZarlicResponse rsp = protocol.unmarshallingResponse(repData); //6、结果处理 if(rsp.getException() != null){ throw rsp.getException(); } return rsp.getReturnValue(); } } }
服务代理类由客户端代理工厂类产生，代理方式是基于Java的动态代理。在处理类ClientInvocationHandler的invoke函数中，定义了一系列的操作，包括获取服务、选择服务提供者、构造请求对象、编组请求对象、网络请求客户端发送请求、解组响应消息、异常处理等。
3.3.2.4 消息协议
复制代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78/** * 消息协议，定义编组请求、解组请求、编组响应、解组响应规范 * */ public interface MessageProtocol { /** * 编组请求 * * @param req 请求信息 * @return 请求字节数组 * @throws Exception 编组请求异常 */ byte[] marshallingRequest(ZarlicRequest req) throws Exception; /** * 解组请求 * * @param data 请求字节数组 * @return 请求信息 * @throws Exception 解组请求异常 */ ZarlicRequest unmarshallingRequest(byte[] data) throws Exception; /** * 编组响应 * * @param rsp 响应信息 * @return 响应字节数组 * @throws Exception 编组响应异常 */ byte[] marshallingResponse(ZarlicResponse rsp) throws Exception; /** * 解组响应 * * @param data 响应字节数组 * @return 响应信息 * @throws Exception 解组响应异常 */ ZarlicResponse unmarshallingResponse(byte[] data) throws Exception; } /** * Java序列化消息协议 * */ public class JavaSerializeMessageProtocol implements MessageProtocol { private byte[] serialize(Object obj) throws Exception { ByteArrayOutputStream bout = new ByteArrayOutputStream(); ObjectOutputStream out = new ObjectOutputStream(bout); out.writeObject(obj); return bout.toByteArray(); } @Override public byte[] marshallingRequest(ZarlicRequest req) throws Exception { return this.serialize(req); } @Override public ZarlicRequest unmarshallingRequest(byte[] data) throws Exception { ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(data)); return (ZarlicRequest) in.readObject(); } @Override public byte[] marshallingResponse(ZarlicResponse rsp) throws Exception { return this.serialize(rsp); } @Override public ZarlicResponse unmarshallingResponse(byte[] data) throws Exception { ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(data)); return (ZarlicResponse) in.readObject(); } }
消息协议主要是定义了客户端如何编组请求、解组响应，服务端如何解组请求、编组响应这四个操作规范。
3.4 服务端编写
3.4.1 服务端需要做什么
首先，服务端要提供远程服务，必须具备服务注册及暴露的能力；在这之后，还需要开启网络服务，供客户端连接。有些项目可能既是服务提供者，又是服务消费者，那什么时候开启服务，什么时候注入服务呢？这里我们引入一个RPC处理者的概念，由它来帮我们开启服务，以及注入服务。
3.4.2 具体实现
服务端的代码结构
服务端做的事情也很简单，注册服务并暴露服务，然后开启网络服务；如果服务端也是消费者，则注入远程服务。
服务注册和服务注入依赖两个自定义注解来实现：
- @Service：注册服务
- @InjectService：注入服务
下面是他们的实现代码：
复制代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40/** * 被该注解标记的服务可提供远程RPC访问的能力 */ @Target({ElementType.TYPE}) @Retention(RetentionPolicy.RUNTIME) @Documented //这个注解只是用来标注生成javadoc的时候是否会被记录。 @Component public @interface Service { String value() default ""; } /** * @Target:注解的作用目标 * * @Target(ElementType.TYPE)——接口、类、枚举、注解 * @Target(ElementType.FIELD)——字段、枚举的常量 * @Target(ElementType.METHOD)——方法 * @Target(ElementType.PARAMETER)——方法参数 * @Target(ElementType.CONSTRUCTOR) ——构造函数 * @Target(ElementType.LOCAL_VARIABLE)——局部变量 * @Target(ElementType.ANNOTATION_TYPE)——注解 * @Target(ElementType.PACKAGE)——包 * * @Retention：注解的保留位置 * * RetentionPolicy.SOURCE:这种类型的Annotations只在源代码级别保留,编译时就会被忽略,在class字节码文件中不包含。 * RetentionPolicy.CLASS:这种类型的Annotations编译时被保留,默认的保留策略,在class文件中存在,但JVM将会忽略,运行时无法获得。 * RetentionPolicy.RUNTIME:这种类型的Annotations将被JVM保留,所以他们能在运行时被JVM或其他使用反射机制的代码所读取和使用。 * @Document：说明该注解将被包含在javadoc中 * @Inherited：说明子类可以继承父类中的该注解 */ /** * 该注解用于注入远程服务 */ @Target({ElementType.FIELD}) @Retention(RetentionPolicy.RUNTIME) @Documented public @interface InjectService { }
3.4.2.1 服务注册（暴露）
复制代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78/** * 服务注册器，定义服务注册规范 */ public interface ServiceRegister { void register(ServiceObject so) throws Exception; ServiceObject getServiceObject(String name) throws Exception; } public class DefaultServiceRegister implements ServiceRegister{ private Map<String,ServiceObject> serviceMap = new HashMap<>(); protected String protocol; protected Integer port; @Override public void register(ServiceObject so) throws Exception { if(so == null){ throw new IllegalArgumentException("Parameter cannot be empty."); } this.serviceMap.put(so.getName,so); } @Override public ServiceObject getServiceObject(String name) throws Exception { return serviceMap.get(name); } } /** Zookeeper服务注册器，提供服务注册、服务暴露的能力 * @author zarlic * @date 2021.12.15 20:34 */ public class ZookeeperExportServiceRegister extends DefaultServiceRegister implements ServiceRegister{ /** * ZK客户端 */ private ZkClient client; public ZookeeperExportServiceRegister(String zkAddress, Integer port, String protocol) { client = new ZkClient(zkAddress); client.setZkSerializer(new ZookeeperSerializer()); this.port = port; this.protocol = protocol; } /** * 服务注册 * @param so * @throws Exception */ @Override public void register(ServiceObject so) throws Exception { super.register(so); Service service = new Service(); String host = InetAddress.getLocalHost().getHostAddress(); String address = host + ":" + port; service.setAddress(address); service.setName(so.getClass().getName()); service.setProtocol(protocol); this.exportService(service); //暴露服务 } private void exportService(Service serviceResource) { String serviceName = serviceResource.getName(); String uri = JSON.toJSONString(serviceResource); try { uri = URLEncoder.encode(uri, ZarlicConstant.UTF_8); } catch (UnsupportedEncodingException e) { e.printStackTrace(); } String servicePath = ZarlicConstant.ZK_SERVICE_PATH + ZarlicConstant.PATH_DELIMITER + serviceName + "/service"; if(!client.exists(servicePath)){ client.createPersistent(servicePath,true); } String uriPath = servicePath + ZarlicConstant.PATH_DELIMITER + uri; if (client.exists(uriPath)) { client.delete(uriPath); } client.createEphemeral(uriPath); //创建临时节点 } }
将指定ServiceObject对象序列化后保存到ZK上，供客户端发现。同时会将服务对象缓存起来，在客户端调用服务时，通过缓存的ServiceObject对象反射指定服务，调用方法。
3.4.2.2 网络服务
复制代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161/** * RPC服务端抽象类 * @author zarlic * @date 2021.12.15 20:34 */ public abstract class RpcServer { /** * 服务端口 */ protected int port; /** * 服务协议 */ protected String protocol; /** * 请求处理者 */ protected RequestHandler handler; public RpcServer(int port, String protocol, RequestHandler handler) { super(); this.port = port; this.protocol = protocol; this.handler = handler; } /** * 开启服务 */ public abstract void start(); /** * 停止服务 */ public abstract void stop(); } /** * Netty RPC服务端，提供Netty网络服务开启、关闭的能力 * @author zarlic * @date 2021.12.15 20:35 */ public class NettyRpcServer extends RpcServer{ private static Logger logger = LoggerFactory.getLogger(NettyRpcServer.class); private Channel channel; public NettyRpcServer(int port, String protocol, RequestHandler handler) { super(port, protocol, handler); } @Override public void start() { NioEventLoopGroup bossGroup = new NioEventLoopGroup(1); NioEventLoopGroup workerGroup = new NioEventLoopGroup(); try { ServerBootstrap b = new ServerBootstrap(); b.group(bossGroup,workerGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG,100) .handler(new LoggingHandler(LogLevel.INFO)).childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline(); pipeline.addLast(new ChannelRequestHandler()); } }); // 启动服务 ChannelFuture f = b.bind(port).sync(); logger.info("Server started successfully."); channel = f.channel(); // 等待服务通道关闭 f.channel().closeFuture().sync(); } catch (InterruptedException e) { e.printStackTrace(); } finally { // 释放线程组资源 bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } } @Override public void stop() { this.channel.close(); } private class ChannelRequestHandler extends ChannelInboundHandlerAdapter{ @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { logger.info("Channel active：{}", ctx); } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { logger.info("The server receives a message: {}", msg); ByteBuf msgBuf = (ByteBuf) msg; byte[] req = new byte[msgBuf.readableBytes()]; msgBuf.readBytes(req); byte[] res = handler.handleRequest(req); logger.info("Send response：{}", msg); ByteBuf respBuf = Unpooled.buffer(res.length); respBuf.writeBytes(res); ctx.write(respBuf); } @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { ctx.flush(); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { cause.printStackTrace(); logger.error("Exception occurred：{}", cause.getMessage()); ctx.close(); } } } /**请求处理者，提供解组请求、编组响应等操作 * @author zarlic * @date 2021.12.15 20:35 */ public class RequestHandler { private MessageProtocol protocol; private ServiceRegister serviceRegister; public RequestHandler(MessageProtocol protocol, ServiceRegister serviceRegister) { super(); this.protocol = protocol; this.serviceRegister = serviceRegister; } public byte[] handleRequest(byte[] data) throws Exception { // 1、解组消息 ZarlicRequest req = this.protocol.unmarshallingRequest(data); // 2、查找服务对象 ServiceObject so = this.serviceRegister.getServiceObject(req.getServiceName()); ZarlicResponse rsp = null; if(so == null){ rsp = new ZarlicResponse(ZarlicStatus.NOT_FOUND); }else { // 3、反射调用对应的过程方法 try { Method method = so.getClazz().getMethod(req.getMethod(), req.getParameterTypes()); Object returnValue = method.invoke(so.getObj(), req.getParameters()); rsp = new ZarlicResponse(ZarlicStatus.SUCCESS); rsp.setReturnValue(returnValue); } catch (NoSuchMethodException | SecurityException | IllegalAccessException | IllegalArgumentException | InvocationTargetException e) { rsp = new ZarlicResponse(ZarlicStatus.ERROR); rsp.setException(e); } } // 4、编组响应消息 return this.protocol.marshallingResponse(rsp); } }
3.4.2.3 RPC处理者
复制代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91/**RPC处理者，支持服务启动暴露、自动注入Service * @author zarlic * @date 2021.12.15 20:33 */ public class DefaultRpcProcessor implements ApplicationListener<ContextRefreshedEvent> { /** * ApplicationContext事件机制是观察者设计模式的实现，通过ApplicationEvent类和ApplicationListener接口，可以实现ApplicationContext事件处理 * 如果容器中有一个ApplicationListener Bean，每当ApplicationContext发布ApplicationEvent时，ApplicationListener Bean将自动被触发。这种事件机制都必须需要程序显示的触发。 * @param event */ @Resource private ClientProxyFactory clientProxyFactory; @Resource private ServiceRegister serviceRegister; @Resource private RpcServer rpcServer; @Override public void onApplicationEvent(ContextRefreshedEvent event) { if(Objects.isNull(event.getApplicationContext().getParent())){ ApplicationContext context = event.getApplicationContext(); // 开启服务 startServer(context); // 注入Service injectService(context); } } private void injectService(ApplicationContext context) { String[] names = context.getBeanDefinitionNames(); for (String name : names) { Class<?> clazz = context.getType(name); if (Objects.isNull(clazz)) continue; Field[] fields = clazz.getDeclaredFields(); for (Field field : fields) { InjectService injectLeisure = field.getAnnotation(InjectService.class); if (Objects.isNull(injectLeisure)) continue; Class<?> fieldClass = field.getType(); Object object = context.getBean(name); field.setAccessible(true); try { field.set(object,clientProxyFactory.getProxy(fieldClass)); } catch (IllegalAccessException e) { e.printStackTrace(); } } } } private void startServer(ApplicationContext context) { Map<String, Object> beans = context.getBeansWithAnnotation(Service.class); if (beans.size() != 0) { boolean startServerFlag = true; for (Object obj : beans.values()) { try { Class<?> clazz = obj.getClass(); Class<?>[] interfaces = clazz.getInterfaces(); ServiceObject so; if(interfaces.length != 1){ Service service = clazz.getAnnotation(Service.class); String value = service.value(); if(value.equals("")){ startServerFlag = false; throw new UnsupportedOperationException("The exposed interface is not specific with '" + obj.getClass().getName() + "'"); } so = new ServiceObject(value,Class.forName(value),obj); }else { Class<?> superClass = interfaces[0]; so = new ServiceObject(superClass.getName(),superClass,obj); } serviceRegister.register(so); } catch (Exception e) { e.printStackTrace(); } } if (startServerFlag) { rpcServer.start(); } } } }
DefaultRpcProcessor实现了ApplicationListener，并监听了ContextRefreshedEvent事件，其效果就是在Spring启动完毕过后会收到一个事件通知，基于这个机制，就可以在这里开启服务，以及注入服务。因为一切已经准备就绪了，所需要的资源都是OK的。
3.5 配置类
复制代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88/** * 参数配置类，实现用户自定义参数 * 在 Spring Boot 项目中，为满足以上要求，我们将大量的参数配置在 application.properties 或 application.yml 文件中， * 通过 @ConfigurationProperties 注解，我们可以方便的获取这些参数值 * @author zarlic * @date 2021.12.15 22:06 */ @EnableConfigurationProperties(ZarlicRpcProperty.class) @ConfigurationProperties("zarlic.rpc") public class ZarlicRpcProperty { /** * 服务注册中心 */ private String registerAddress = "127.0.0.1:2181"; /** * 服务暴露端口 */ private Integer serverPort = 19000; /** * 服务协议 */ private String protocol = "zarlic"; public String getRegisterAddress() { return registerAddress; } public void setRegisterAddress(String registerAddress) { this.registerAddress = registerAddress; } public Integer getServerPort() { return serverPort; } public void setServerPort(Integer serverPort) { this.serverPort = serverPort; } public String getProtocol() { return protocol; } public void setProtocol(String protocol) { this.protocol = protocol; } } @Configuration public class AutoConfiguration { @Bean public DefaultServiceRegister defaultServiceRegister(){ return new DefaultServiceRegister(); } @Bean public ClientProxyFactory clientProxyFactory(@Autowired ZarlicRpcProperty zarlicRpcProperty){ ClientProxyFactory clientProxyFactory = new ClientProxyFactory(); //设置服务发现者 clientProxyFactory.setServiceDiscoverer(new ZookeeperServiceDiscoverer(zarlicRpcProperty.getRegisterAddress())); //设置支持的协议 Map<String, MessageProtocol> supportMessageProtocols = new HashMap<>(); supportMessageProtocols.put(zarlicRpcProperty.getProtocol(),new JavaSerializeMessageProtocol()); clientProxyFactory.setSupportMessageProtocols(supportMessageProtocols); //设置网络层的实现 clientProxyFactory.setNetClient(new NettyNetClient()); return clientProxyFactory; } @Bean public ServiceRegister serviceRegister(@Autowired ZarlicRpcProperty zarlicRpcProperty){ return new ZookeeperExportServiceRegister(zarlicRpcProperty.getRegisterAddress(), zarlicRpcProperty.getServerPort(),zarlicRpcProperty.getProtocol()); } @Bean public RequestHandler requestHandler(@Autowired ServiceRegister serviceRegister){ return new RequestHandler(new JavaSerializeMessageProtocol(),serviceRegister); } @Bean public RpcServer rpcServer(@Autowired RequestHandler requestHandler,@Autowired ZarlicRpcProperty zarlicRpcProperty){ return new NettyRpcServer(zarlicRpcProperty.getServerPort(), zarlicRpcProperty.getProtocol(),requestHandler); } @Bean public ZarlicRpcProperty zarlicRpcProperty(){ return new ZarlicRpcProperty(); } }
最后
以上就是友好巨人最近收集整理的关于手写一个RPC框架（造轮子）1 涉及的知识点2 RPC 基础知识3 手写RPC的全部内容，更多相关手写一个RPC框架（造轮子）1内容请搜索靠谱客的其他文章。
本图文内容来源于网友提供，作为学习参考使用，或来自网络收集整理，版权属于原作者所有。
发表评论 取消回复