1.简单的实现

https://blog.csdn.net/qq_40856284/article/details/106972652

本章对应的commit为73aa960,完整项目为https://github.com/CN-GuoZiyang/My-RPC-Framework/tree/73aa960b0c457770859f81a3210de56370862439

思路

用(抄)一下Guide哥的一张图:

RPC框架思路

那么我们首先要思考,RPC框架的原理。

原理很简单,客户端和服务端都可以访问到通用的接口,但是只有服务端有这个接口的实现类,客户端调用这个接口的方式,是通过网络传输,告诉服务端我要调用这个接口,服务端收到之后找到这个接口的实现类,并且执行,将执行的结果返回给客户端,作为客户端调用接口方法的返回值。

原理很简单,但是实现值得商榷,例如客户端怎么知道服务端的地址?客户端怎么告诉服务端我要调用的接口?客户端怎么传递参数?只有接口客户端怎么生成实现类……等等等等。

这一章,我们就来探讨一个最简单的实现。一个最简单的实现,基于这样一个假设,那就是客户端已经知道了服务端的地址,这部分会由后续的服务发现机制完善。

通用接口

我们先把通用的接口写好,然后再来看怎么实现客户端和服务端。

接口如下:

1
2
3
4
public interface HelloService {
String hello(HelloObject object);
}
123

hello方法需要传递一个对象,HelloObject对象,定义如下:

1
2
3
4
5
6
7
@Data
@AllArgsConstructor
public class HelloObject implements Serializable {
private Integer id;
private String message;
}
123456

注意这个对象需要实现Serializable接口,因为它需要在调用过程中从客户端传递给服务端。

接着我们在服务端对这个接口进行实现,实现的方式也很简单,返回一个字符串就行:

1
2
3
4
5
6
7
8
public class HelloServiceImpl implements HelloService {
private static final Logger logger = LoggerFactory.getLogger(HelloServiceImpl.class);
@Override
public String hello(HelloObject object) {
logger.info("接收到:{}", object.getMessage());
return "这是掉用的返回值,id=" + object.getId();
}
}

传输协议

严格来说,这并不能算是协议……但也大致算一个传输格式吧。

我们来思考一下,服务端需要哪些信息,才能唯一确定服务端需要调用的接口的方法呢?

首先,就是接口的名字,和方法的名字,但是由于方法重载的缘故,我们还需要这个方法的所有参数的类型,最后,客户端调用时,还需要传递参数的实际值,那么服务端知道以上四个条件,就可以找到这个方法并且调用了。我们把这四个条件写到一个对象里,到时候传输时传输这个对象就行了。即RpcRequest对象:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
@Data
@Builder
public class RpcRequest implements Serializable {
/**
* 待调用接口名称
*/
private String interfaceName;
/**
* 待调用方法名称
*/
private String methodName;
/**
* 调用方法的参数
*/
private Object[] parameters;
/**
* 调用方法的参数类型
*/
private Class<?>[] paramTypes;
}

参数类型我是直接使用Class对象,其实用字符串也是可以的。

那么服务器调用完这个方法后,需要给客户端返回哪些信息呢?如果调用成功的话,显然需要返回值,如果调用失败了,就需要失败的信息,这里封装成一个RpcResponse对象:

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
@Data
public class RpcResponse<T> implements Serializable {
/**
* 响应状态码
*/
private Integer statusCode;
/**
* 响应状态补充信息
*/
private String message;
/**
* 响应数据
*/
private T data;

public static <T> RpcResponse<T> success(T data) {
RpcResponse<T> response = new RpcResponse<>();
response.setStatusCode(ResponseCode.SUCCESS.getCode());
response.setData(data);
return response;
}
public static <T> RpcResponse<T> fail(ResponseCode code) {
RpcResponse<T> response = new RpcResponse<>();
response.setStatusCode(code.getCode());
response.setMessage(code.getMessage());
return response;
}
}

这里还多写了两个静态方法,用于快速生成成功与失败的响应对象。其中,statusCode属性可以自行定义,客户端服务端一致即可。

客户端的实现——动态代理

客户端方面,由于在客户端这一侧我们并没有接口的具体实现类,就没有办法直接生成实例对象。这时,我们可以通过动态代理的方式生成实例,并且调用方法时生成需要的RpcRequest对象并且发送给服务端。

这里我们采用JDK动态代理,代理类是需要实现InvocationHandler接口的。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
public class RpcClientProxy implements InvocationHandler {
private String host;
private int port;

public RpcClientProxy(String host, int port) {
this.host = host;
this.port = port;
}

@SuppressWarnings("unchecked")
public <T> T getProxy(Class<T> clazz) {
return (T) Proxy.newProxyInstance(clazz.getClassLoader(), new Class<?>[]{clazz}, this);
}
}

我们需要传递host和port来指明服务端的位置。并且使用getProxy()方法来生成代理对象。

InvocationHandler接口需要实现invoke()方法,来指明代理对象的方法被调用时的动作。在这里,我们显然就需要生成一个RpcRequest对象,发送出去,然后返回从服务端接收到的结果即可:

1
2
3
4
5
6
7
8
9
10
11
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
RpcRequest rpcRequest = RpcRequest.builder()
.interfaceName(method.getDeclaringClass().getName())
.methodName(method.getName())
.parameters(args)
.paramTypes(method.getParameterTypes())
.build();
RpcClient rpcClient = new RpcClient();
return ((RpcResponse) rpcClient.sendRequest(rpcRequest, host, port)).getData();
}

生成RpcRequest很简单,我使用Builder模式来生成这个对象。发送的逻辑我使用了一个RpcClient对象来实现,这个对象的作用,就是将一个对象发过去,并且接受返回的对象。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
public class RpcClient {

private static final Logger logger = LoggerFactory.getLogger(RpcClient.class);

public Object sendRequest(RpcRequest rpcRequest, String host, int port) {
try (Socket socket = new Socket(host, port)) {
ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream());
ObjectInputStream objectInputStream = new ObjectInputStream(socket.getInputStream());
objectOutputStream.writeObject(rpcRequest);
objectOutputStream.flush();
return objectInputStream.readObject();
} catch (IOException | ClassNotFoundException e) {
logger.error("调用时有错误发生:", e);
return null;
}
}
}

我的实现很简单,直接使用Java的序列化方式,通过Socket传输。创建一个Socket,获取ObjectOutputStream对象,然后把需要发送的对象传进去即可,接收时获取ObjectInputStream对象,readObject()方法就可以获得一个返回的对象。

服务端的实现——反射调用

服务端的实现就简单多了,使用一个ServerSocket监听某个端口,循环接收连接请求,如果发来了请求就创建一个线程,在新线程中处理调用。这里创建线程采用线程池:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
public class RpcServer {

private final ExecutorService threadPool;
private static final Logger logger = LoggerFactory.getLogger(RpcServer.class);

public RpcServer() {
int corePoolSize = 5;
int maximumPoolSize = 50;
long keepAliveTime = 60;
BlockingQueue<Runnable> workingQueue = new ArrayBlockingQueue<>(100);
ThreadFactory threadFactory = Executors.defaultThreadFactory();
threadPool = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, TimeUnit.SECONDS, workingQueue, threadFactory);
}

}

这里简化了一下,RpcServer暂时只能注册一个接口,即对外提供一个接口的调用服务,添加register方法,在注册完一个服务后立刻开始监听:

1
2
3
4
5
6
7
8
9
10
11
12
public void register(Object service, int port) {
try (ServerSocket serverSocket = new ServerSocket(port)) {
logger.info("服务器正在启动...");
Socket socket;
while((socket = serverSocket.accept()) != null) {
logger.info("客户端连接!Ip为:" + socket.getInetAddress());
threadPool.execute(new WorkerThread(socket, service));
}
} catch (IOException e) {
logger.error("连接时有错误发生:", e);
}
}

这里向工作线程WorkerThread传入了socket和用于服务端实例service。

WorkerThread实现了Runnable接口,用于接收RpcRequest对象,解析并且调用,生成RpcResponse对象并传输回去。run方法如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
@Override
public void run() {
try (ObjectInputStream objectInputStream = new ObjectInputStream(socket.getInputStream());
ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream())) {
RpcRequest rpcRequest = (RpcRequest) objectInputStream.readObject();
Method method = service.getClass().getMethod(rpcRequest.getMethodName(), rpcRequest.getParamTypes());
Object returnObject = method.invoke(service, rpcRequest.getParameters());
objectOutputStream.writeObject(RpcResponse.success(returnObject));
objectOutputStream.flush();
} catch (IOException | ClassNotFoundException | NoSuchMethodException | IllegalAccessException | InvocationTargetException e) {
logger.error("调用或发送时有错误发生:", e);
}
}

其中,通过class.getMethod方法,传入方法名和方法参数类型即可获得Method对象。如果你上面RpcRequest中使用String数组来存储方法参数类型的话,这里你就需要通过反射生成对应的Class数组了。通过method.invoke方法,传入对象实例和参数,即可调用并且获得返回值。

测试

服务端侧,我们已经在上面实现了一个HelloService的实现类HelloServiceImpl的实现类了,我们只需要创建一个RpcServer并且把这个实现类注册进去就行了:

1
2
3
4
5
6
7
public class TestServer {
public static void main(String[] args) {
HelloService helloService = new HelloServiceImpl();
RpcServer rpcServer = new RpcServer();
rpcServer.register(helloService, 9000);
}
}

服务端开放在9000端口。

客户端方面,我们需要通过动态代理,生成代理对象,并且调用,动态代理会自动帮我们向服务端发送请求的:

1
2
3
4
5
6
7
8
9
public class TestClient {
public static void main(String[] args) {
RpcClientProxy proxy = new RpcClientProxy("127.0.0.1", 9000);
HelloService helloService = proxy.getProxy(HelloService.class);
HelloObject object = new HelloObject(12, "This is a message");
String res = helloService.hello(object);
System.out.println(res);
}
}

我们这里生成了一个HelloObject对象作为方法的参数。

首先启动服务端,再启动客户端,服务端输出:

1
2
3
4
服务器正在启动...
客户端连接!Ip为:127.0.0.1
接收到:This is a message
123

客户端输出:

1
这是调用的返回值,id=12