实现思路
客户端把需要调用的接口信息封装到RpcRequest对象中，通过socket通信把对象发送到接口提供方，服务端收到RpcRequest对象后，通过反射的方式进行方法调用，返回调用结果到客户端。-
工程结构：
截屏2020-04-11 下午5.44.48.png
截屏2020-04-11 下午5.45.26.png 工程功能：
rpc-client：客户端，调用rpc-server中提供的接口
rpc-server：服务端，创建两个模块rpc-server-api、rpc-server-service
rpc-server-api：定义接口和公共的类。
rpc-server-service：接口实现和接口实现类的发布。
-
RpcRequest类图：
RpcRequest.png
- 服务端实现过程
在rpc-server-api中声明ISayHello接口和RpcRequest类。
public interface ISayHello {
public String sayHello(String content);
}
//封装请求中要调用的方法，通过这个对象的实例可以反射通过找到并调用任意一个类的方法
public class RpcRequest implements Serializable {
private String className;
private String methodName;
private Object [] parameters;
private Class [] types;
public String getClassName() {
return className;
}
public void setClassName(String className) {
this.className = className;
}
public String getMethodName() {
return methodName;
}
public void setMethodName(String methodName) {
this.methodName = methodName;
}
public Object[] getParameters() {
return parameters;
}
public void setParameters(Object[] parameters) {
this.parameters = parameters;
}
public Class[] getTypes() {
return types;
}
public void setTypes(Class[] types) {
this.types = types;
}
}
- 在rpc-server-service中实现并发布接口
public class SayHelloImpl implements ISayHello {
@Override
public String sayHello(String content) {
return "The Content is " + content;
}
}
/**
* 使用反射实现方法调用
*/
public class ProcessHandler implements Runnable {
Socket socket ;
Object service;
public ProcessHandler(Socket socket, Object service) {
this.socket = socket;
this.service = service;
}
//这里真正的去使用反射调用请求进来的接口的方法
@Override
public void run() {
ObjectInputStream inputStream = null;
ObjectOutputStream outputStream = null;
try {
inputStream = new ObjectInputStream(socket.getInputStream());
//获取请求进来要调用的方法
RpcRequest request = (RpcRequest) inputStream.readObject();
System.out.println("接收到请求");
Object result = invoke(request);
//返回调用结果
outputStream = new ObjectOutputStream(socket.getOutputStream());
outputStream.writeObject(result);
outputStream.flush();
} catch (Exception e) {
e.printStackTrace();
}finally {
try {
outputStream.close();
inputStream.close();
socket.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
/**
* 通过反射的方式找到要调用的类名，并调用方法
* @param request
* @return
* @throws Exception
*/
public Object invoke(RpcRequest request) throws Exception {
Class clazz = Class.forName(request.getClassName());
Method method = clazz.getDeclaredMethod(request.getMethodName(),request.getTypes());
Object result = method.invoke(service,request.getParameters());
return result;
}
}
/**
* socket服务端，用于发布接口的实现类
*
*/
public class RpcProxyServer {
//线程池，socket连接之后把请求丢给其他线程去处理，避免阻塞
private ExecutorService executorService = Executors.newCachedThreadPool();
//用于发布服务
public void publisher(Object service, int port) {
ServerSocket serverSocket = null;
Socket socket = null;
try {
serverSocket = new ServerSocket(port);
//把调用请求丢到线程池中处理
while (true) {
socket = serverSocket.accept();
System.out.println("接收到来自" + socket.getPort() +"的连接");
executorService.execute(new ProcessHandler(socket, service));
}
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
socket.close();
serverSocket.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
public class App
{
public static void main( String[] args )
{
/**
* 对象实例的发布,这里只有一个接口需要发布
* 当有多个接口实例需要发布的时候，可以先把所有的实例都注册到容器中
* 然后遍历容器，依次发布
* 端口号8888可以看作是rpc框架通讯的默认端口号
*/
ISayHello sayHello = new SayHelloImpl();
RpcProxyServer rpcProxyServer = new RpcProxyServer();
rpcProxyServer.publisher(sayHello, 8888);
}
}
- rpc-client端调用实现：
因为rpc-server-api是公共模块，所以在客户端是可以获取到ISayHello接口的，但是获取不到对应的实现类，这时就可以使用rpc请求的方式调用服务端的接口实现。
这个时候如果只是单纯的远程调用一个接口，直接写一个socket请求过去是可以的，如果是多个接口的调用，可能会想到使用封装的方式抽象出一个方法。这里使用动态代理的方式去做远程调用，使得调用的过程中更具有扩展性。
- 声明一个代理方法
//动态代理
public class RpcProxyClient {
/**
*
* @param host 请求的主机
* @param port 端口
* @param proxyedClass 被代理的类
* @param <T>
* @return 返回被代理的类的实例
*/
public <T> T clientProxy(String host , int port , Class<T> proxyedClass){
return (T)Proxy.newProxyInstance(proxyedClass.getClassLoader(), new Class[]{proxyedClass}, new RpcInvocationHandler(host,port));
}
}
- 封装请求和返回结果
public class RpcInvocationHandler implements InvocationHandler {
private String host;
private int port;
public RpcInvocationHandler(String host, int port) {
this.host = host;
this.port = port;
}
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
System.out.println("开始进行代理");
//封装请求
RpcRequest request = new RpcRequest();
//request.setClassName(proxy.getClass().getName());
request.setClassName(method.getDeclaringClass().getName());
request.setMethodName(method.getName());
request.setParameters(args);
request.setTypes(method.getParameterTypes());
//远程调用
RpcNetTransport netTransport = new RpcNetTransport(host,port);
Object result = netTransport.send(request);
return result;
}
}
- 真正的远程调用
public class RpcNetTransport {
private String host;
private int port;
public RpcNetTransport(String host, int port) {
this.host = host;
this.port = port;
}
//远程调用
public Object send(RpcRequest request){
ObjectInputStream inputStream = null;
ObjectOutputStream outputStream = null;
try {
Socket socket = new Socket(host,port);
outputStream = new ObjectOutputStream(socket.getOutputStream());
outputStream.writeObject(request);
outputStream.flush();
inputStream = new ObjectInputStream(socket.getInputStream());
Object result = inputStream.readObject();
return result;
} catch (Exception e) {
e.printStackTrace();
}finally {
try {
outputStream.close();
inputStream.close();
}catch(Exception e){
e.printStackTrace();
}
}
return null;
}
}
- 测试
public class App {
public static void main( String[] args )
{
Thread t1 = new Thread(()->{
RpcProxyClient client = new RpcProxyClient();
ISayHello iSayHello = client.clientProxy("localhost",8888,ISayHello.class);
String result = iSayHello.sayHello("Tomas");
System.out.println(result);
},"Thread-1");
Thread t2 = new Thread(()->{
RpcProxyClient client = new RpcProxyClient();
ISayHello iSayHello = client.clientProxy("localhost",8888,ISayHello.class);
String result = iSayHello.sayHello("Tomas");
System.out.println(result);
},"Thread-2");
Thread t3 = new Thread(()->{
RpcProxyClient client = new RpcProxyClient();
ISayHello iSayHello = client.clientProxy("localhost",8888,ISayHello.class);
String result = iSayHello.sayHello("Tomas");
System.out.println(result);
},"Thread-3");
t1.start();
t2.start();
t3.start();
}
}
-
总结
api作为公共模块用于定义接口个其他的公共类；客户端封装请求之后，通过socket实现远程调用；客户端收到socket消息之后进行处理，完成方法调用后返回处理结果。
总体架构.png
