RPC框架实现之路01-极简的RPC框架实现

2021-03-06 本文已影响0人 czczccz

Version 01

动态代理通过jdk动态代理方式实现，网络通信通过Socket实现，服务注册和获取通过HashMap实现，对象的序列化通过jdk 序列化方式实现。
此版本的代码在：Version 01代码
RPC框架会逐步进行完善，目前是简单的第一版本。

Module：rpc-framework-common

用于搭载请求信息和失败信息。

dto包

RpcRequest

此模块主要编写rpcClient和rpcServer之间网络传输时候的媒介：RpcRequest

其作为接口名、方法名、参数、参数类型的载体。

因为要通过网络进行传输，所以需要实现序列化接口。

这里使用lombok注解了@Builder和@Data
RpcResponse

用来做响应的RpcResponse，主要搭载RpcServer向RpcClient返回的信息，包括状态码，状态信息以及返回的数据。

提供两个状态的设置数据的方法，success传参为返回的数据，因为状态码和状态信息均为成功。

fail需要传入枚举类RpcResponseCode，因为失败的原因是有多种的，需要由上游决定。

enumeration包

枚举类型

RpcResponseCode

枚举类型。为响应状态码，与RpcResponse协同工作
RpcErrorMessageEnum

枚举类型。为异常信息，与RpcException协同工作

exception包

RpcException

自定义异常，使用到了RpcErrorMessageEnum

Module：rpc-framework-simple

此模块主要编写RPC客户端和RPC服务端

registry包

ServiceRegistry

定义接口，其功能为服务的获取以及服务的注册。

注册服务需要传入指定的服务；

获取服务需要传入服务名。

DefaultServiceRegistry

实现了上述接口

在注册服务中，服务名为接口名，比如com.wyb.neu.HelloServiceImpl，并且保存需要注册的服务是通过HashMap来保存的，HashMap的key为接口名，value为实现的类。

添加服务的过程会将当前服务实现的所有接口都添加进去。

 @Override
 public <T> void register(T service) {
 //获取服务名
 String serviceName = service.getClass().getCanonicalName();
 //是否已经注册过
 if (registeredService.contains(serviceName)) {
 return;
 }
 //先添加到注册过的set集合中
 registeredService.add(serviceName);
 //获取接口集合
 Class[] interfaces = service.getClass().getInterfaces();
 //如果接口集合是0
 if (interfaces.length == 0) {
 throw new RpcException(RpcErrorMessageEnum.SERVICE_NOT_IMPLEMENT_ANY_INTERFACE);
 }
 //将所有接口都添加进去
 for (Class i : interfaces) {
 serviceMap.put(i.getCanonicalName(), service);
 }
 logger.info("Add service: {} and interfaces:{}", serviceName, service.getClass().getInterfaces());
 }```

那么是如何获取服务的呢？

传入服务名，判断HashMap里面有没有就可以了。

 ```java
 @Override
 public Object getService(String serviceName) {
 Object service = serviceMap.get(serviceName);
 if(null == service)
 throw new RpcException(RpcErrorMessageEnum.SERVICE_CAN_NOT_BE_FOUND);
 return service;
 }

remoting包

RpcClient

使用jdk seriable以及Socket完成RpcRequest的发送，并且通过Socket以及序列化再接收返回信息，进行判断。

主要代码：

  public Object sendRpcRequest(RpcRequest rpcRequest, String host, int port)
     {
     //创建socket对象，根据指定的主机和端口
     try(Socket socket = new Socket(host,port)){
     ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream());
     objectOutputStream.writeObject(rpcRequest);
     ObjectInputStream objectInputStream = new ObjectInputStream(socket.getInputStream());
    //            return objectInputStream.readObject();
    //            Server端的信息都封装到了RpcServer中，后续操作是获取RpcResponse，然后判断是否正确执行
     RpcResponse rpcResponse = (RpcResponse)objectInputStream.readObject();
     if(rpcResponse == null)
     {
     logger.error("Invoke Service Fail , Service Name is : {}",rpcRequest.getInterfaceName());
     }
     if (rpcResponse.getCode() == null || !rpcResponse.getCode().equals(RpcResponseCode.SUCCESS.getCode())) {
     logger.error("调用服务失败,serviceName:{},RpcResponse:{}", rpcRequest.getInterfaceName(), rpcResponse);
     throw new RpcException(RpcErrorMessageEnum.SERVICE_INVOCATION_FAILURE, "interfaceName:" + rpcRequest.getInterfaceName());
     }
     return rpcResponse.getData();
     }catch (IOException | ClassNotFoundException e)
     {
     logger.error("occur exception ", e);
     }
     return null;
     }

RpcClientProxy

使用jdk动态代理，jdkProxy

主要代码：

 public <T>T getProxy(Class<T> clazz){
  // 参数1：被代理类的类加载器 参数2:被代理类的接口
  return (T) Proxy.newProxyInstance(clazz.getClassLoader(), new Class<?>[]{clazz}, this);
 }
 @Override
 public Object invoke(Object proxy, Method method, Object[] args)  {
  logger.info("Call \"invoke method\" and invoked method: {}", method.getName());
  RpcRequest rpcRequest = RpcRequest.builder().methodName(method.getName())
  .parameters(args)
  .interfaceName(method.getDeclaringClass().getName())
  .paramTypes(method.getParameterTypes())
  .build();
  RpcClient rpcClient = new RpcClient();
  return rpcClient.sendRpcRequest(rpcRequest,host,port);
 }

RpcServer

主要使用到ExecutorService线程池、Socket

主要功能为注册服务，并且不断监听客户端，如果收到客户端的请求，将请求执行的任务加入到线程池中执行。

主要的实现逻辑：

  public void start(int port) {

   try (ServerSocket server = new ServerSocket(port);) {
   logger.info("server starts...");
   Socket socket;
   while ((socket = server.accept()) != null) {
   logger.info("client connected");
   threadPool.execute(new RpcRequestHandlerRunnable(socket, rpcRequestHandler, serviceRegistry));
   }
   threadPool.shutdown();
   } catch (IOException e) {
   logger.error("occur IOException:", e);
   }
  }

RpcRequestHandlerRunnable

实现了Runnable接口，主要编写线程执行的主体run方法，其里面完成服务的获取，并调用RpcRequestHandler完成方法的执行，并将结果写回

@Override
public void run() {
try (ObjectInputStream objectInputStream = new ObjectInputStream(socket.getInputStream());
ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream())) {
RpcRequest rpcRequest = (RpcRequest) objectInputStream.readObject();//获取Client传入的RpcRequest
String interfaceName = rpcRequest.getInterfaceName();//获取要执行的接口名
Object service = serviceRegistry.getService(interfaceName);//根据接口名去注册中心查找Service
Object result = rpcRequestHandler.handle(rpcRequest, service);//交到RpcRequestHandler中执行
objectOutputStream.writeObject(RpcResponse.success(result));
objectOutputStream.flush();
} catch (IOException | ClassNotFoundException e) {
logger.error("occur exception:", e);
}
}

RpcRequestHandler

利用反射，完成方法的调用，所执行的方法再rpcRequest中保存着，service的话，在线程里面已经完成了获取

 public Object handle(RpcRequest rpcRequest, Object service) {
 Object result = null;
 try {
 result = invokeTargetMethod(rpcRequest, service);
 logger.info("service:{} successful invoke method:{}", rpcRequest.getInterfaceName(), rpcRequest.getMethodName());
 } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) {
 logger.error("occur exception", e);
 }
 return result;
 }

 private Object invokeTargetMethod(RpcRequest rpcRequest, Object service) throws NoSuchMethodException, IllegalAccessException, InvocationTargetException {
 Method method = service.getClass().getMethod(rpcRequest.getMethodName(), rpcRequest.getParamTypes());
 if (null == method) {
 return RpcResponse.fail(RpcResponseCode.NOT_FOUND_METHOD);
 }
 return method.invoke(service, rpcRequest.getParameters());
 }

Module:hello-service-api

HelloService

接口，服务器提供的方法接口，服务器提供的服务

public interface HelloService {
 String hello(Hello hello);
}

Hello

实现了序列化，因为这个类也需要进行网络传输，是HelloService的参数

public class Hello implements Serializable {
    private String message;
    private String description;
    //get & set

    public Hello(String message, String description) {
        this.message = message;
        this.description = description;
    }
}

example-server

HelloServiceImpl

实现了HelloService中的Hello方法，服务器中的实现类

public class HelloServiceImpl implements HelloService{
    private static final Logger logger = LoggerFactory.getLogger(HelloServiceImpl.class);

    public String sayHello(Hello hello) {
        logger.info("HelloServiceImpl Get: {}",hello.getMessage());
        String results = "Hello description is "+hello.getDescription();
        logger.info("HelloServiceImpl Return: {}",results);
        return results;
    }
}

RpcFrameworkSimpleServerMain

启动服务器，注册服务，传入需要注册的服务以及端口号

public class RpcFrameworkSimpleMain {
    public static void main(String[] args) {
        HelloServiceImpl helloService = new HelloServiceImpl();
        DefaultServiceRegistry defaultServiceRegistry = new DefaultServiceRegistry();
        // 手动注册
        defaultServiceRegistry.register(helloService);
        RpcServer rpcServer = new RpcServer(defaultServiceRegistry);
        rpcServer.start(9999);
    }
}

example-client

RpcFrameworkSimpleClientMain

启动客户端，获取服务

public class RpcFrameworkSimpleClientMain {
    public static void main(String[] args) {
        RpcClientProxy rpcClientProxy = new RpcClientProxy("127.0.0.1",9999);
        HelloService helloService = rpcClientProxy.getProxy(HelloService.class);
        String hello = helloService.sayHello(new Hello("Hello","Hello World"));
        System.out.println(hello);
    }
}

运行实例

1：启动Server

Start Server

2：启动Client

此时的Client：

Start Client

此时的Server：

Server when Start Client