Myrpc version 2

RPC The concept of

Background knowledge

• RPC Basic concepts of , Core functions

common RPC frame

Duboo Basic function

1. Telematics
2. Transparent remote procedure call based on interface method
3. Load balancing
4. Service registry

RPC The process

client Call the remote method -> request serialize -> Protocol code -> Network transmission -> Server side -> Deserialization request -> Call the local method to get response -> serialize -> code ->……

---

Version iteration process

Catalog

from 0 At the beginning RPC Iterative process of ：

• version0 edition ： Complete a with less than 100 lines of code RPC Example
• version1 edition ： Improve the common message format （request,response）, The client's dynamic proxy completes the request Encapsulation of message formats
• version2 edition ： Support the server to expose multiple service interfaces , Server program abstraction , Normalization
• version3 edition ： Use a high-performance network framework netty The realization of network communication , And the refactoring of client code
• version4 edition ： Custom message format , Support multiple serialization methods （java Native , json…）
• version5 edition : Implementation of server registration and discovery ,zookeeper As a registry
• version6 edition : Implementation of load balancing strategy

---

2.MyRPC edition 2

Background knowledge

• Code decoupling
• Thread pool

Questions in this section ：

If a server needs to provide interfaces for multiple services , For example, add a new BlogService, How to deal with ?

//  It's natural to think of using a Map To preserve ,<interfaceName, xxxServiceImpl>
UserService userService = new UserServiceImpl();
BlogService blogService = new BlogServiceImpl();
Map<String, Object>.put("***.userService", userService);
Map<String, Object>.put("***.blogService", blogService);
//  Here comes a request, We can go from map Get the corresponding service from the 
Object service = map.get(request.getInterfaceName())

Version upgrade process

Work before update ： Update a new service interface sample and pojo class

//  New service interface 
public interface BlogService {
    
    Blog getBlogById(Integer id);
}
//  The new service interface implementation class of the server 
public class BlogServiceImpl implements BlogService {
    
    @Override
    public Blog getBlogById(Integer id) {
    
        Blog blog = Blog.builder().id(id).title(" My blog ").useId(22).build();
        System.out.println(" The client inquired "+id+" Blog ");
        return blog;
    }
}
// pojo class 
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class Blog implements Serializable {
    
    private Integer id;
    private Integer useId;
    private String title;
}

to update 1： HashMap<String, Object> Add implementation classes for multiple services

public class TestServer {
    
    public static void main(String[] args) {
    
        UserService userService = new UserServiceImpl();
        BlogService blogService = new BlogServiceImpl();
        Map<String, Object> serviceProvide = new HashMap<>();
        //  Expose two service interfaces ,  That is to say RPCServer To add a HashMap
        serviceProvide.put("com.ganghuan.myRPCVersion2.service.UserService",userService);
        serviceProvide.put("com.ganghuan.myRPCVersion2.service.BlogService",blogService);
        RPCServer RPCServer = new SimpleRPCRPCServer(serviceProvide);
        RPCServer.start(8899);
    }
}
//  The details of the implementation are not discussed here , Because it should also be optimized , Let's start by loosely coupling the server code , Come back to discuss 

to update 2： Server side code refactoring

1. abstract RPCServer, Open and closed principle

//  hold RPCServer Abstract to interface , In the future, the server can implement this interface 
public interface RPCServer {
    
    void start(int port);
    void stop();
}

2. RPCService Implementation of simple version

/** *  This implementation class represents java The original BIO Monitor mode , A mission , Just new A thread to handle  *  For task handling, see WorkThread in  */
public class SimpleRPCRPCServer implements RPCServer {
    
    //  Save the service interface name -> service Object's map
    private Map<String, Object> serviceProvide;
    public SimpleRPCRPCServer(Map<String,Object> serviceProvide){
    
        this.serviceProvide = serviceProvide;
    }
    public void start(int port) {
    
        try {
    
            ServerSocket serverSocket = new ServerSocket(port);
            System.out.println(" The server has started ");
            // BIO Mode of monitoring Socket
            while (true){
    
                Socket socket = serverSocket.accept();
                //  Start a new thread to process 
                new Thread(new WorkThread(socket,serviceProvide)).start();
            }
        } catch (IOException e) {
    
            e.printStackTrace();
            System.out.println(" Server start failed ");
        }
    }
    public void stop(){
    
    }
}

3. To enhance performance , We also provide Thread pool version The implementation of the

public class ThreadPoolRPCRPCServer implements RPCServer {
    
    private final ThreadPoolExecutor threadPool;
    private Map<String, Object> serviceProvide;
    public ThreadPoolRPCRPCServer(Map<String, Object> serviceProvide){
    
        threadPool = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(),
                1000, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(100));
        this.serviceProvide = serviceProvide;
    }
    public ThreadPoolRPCRPCServer(Map<String, Object> serviceProvide, int corePoolSize,
                                  int maximumPoolSize,
                                  long keepAliveTime,
                                  TimeUnit unit,
                                  BlockingQueue<Runnable> workQueue){
    
        
        threadPool = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
        this.serviceProvide = serviceProvide;
    }
    @Override
    public void start(int port) {
    
        System.out.println(" The server has started ");
        try {
    
            ServerSocket serverSocket = new ServerSocket(port);
            while(true){
    
                Socket socket = serverSocket.accept();
                threadPool.execute(new WorkThread(socket,serviceProvide));
            }
        } catch (IOException e) {
    
            e.printStackTrace();
        }
    }
    @Override
    public void stop() {
    
    }
}

4. Work task class , Separate from the server code , Simplify the server code , Principle of single responsibility

/** *  Here we are responsible for parsing the request request , Execute service methods , Return to the client  * 1.  from request obtain interfaceName 2.  according to interfaceName stay serviceProvide Map Get the implementation class of the server in  * 3.  from request Get the method name in , Parameters ,  Using reflection to execute methods in services  4.  Get the results , Encapsulated into response, write in socket */
@AllArgsConstructor
public class WorkThread implements Runnable{
    
    private Socket socket;
    private Map<String, Object> serviceProvide;
    @Override
    public void run() {
    
        try {
    
            ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream());
            ObjectInputStream ois = new ObjectInputStream(socket.getInputStream());
            //  Read from the client request
            RPCRequest request = (RPCRequest) ois.readObject();
            //  Reflection calls the service method to get the return value 
            RPCResponse response = getResponse(request);
            // Write to client 
            oos.writeObject(response);
            oos.flush();
        }catch (IOException | ClassNotFoundException e){
    
            e.printStackTrace();
            System.out.println(" from IO Error reading data in ");
        }
    }
    private RPCResponse getResponse(RPCRequest request){
    
        //  Get the service name 
        String interfaceName = request.getInterfaceName();
        //  Get the corresponding service implementation class of the server 
        Object service = serviceProvide.get(interfaceName);
        //  Reflection calls method 
        Method method = null;
        try {
    
            method = service.getClass().getMethod(request.getMethodName(), request.getParamsTypes());
            Object invoke = method.invoke(service, request.getParams());
            return RPCResponse.success(invoke);
        } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) {
    
            e.printStackTrace();
            System.out.println(" Method execution error ");
            return RPCResponse.fail();
        }
    }
}

The server code is reconstructed for the first time .

to update 3： Service exposure class , Here is the update 1**, We found that the service interface name is us ** Direct handwritten , You can actually use class Object gets automatically

/** *  Previously used here Map Simple implementation  *  Store the service interface name and the implementation class corresponding to the server  *  When a service is started, its related implementation classes should be exposed 0 *  according to request Medium interface Call the relevant implementation classes in the server  */
public class ServiceProvider {
    
    /** *  An implementation class may implement multiple interfaces  */
    private Map<String, Object> interfaceProvider;
    public ServiceProvider(){
    
        this.interfaceProvider = new HashMap<>();
    }
    public void provideServiceInterface(Object service){
    
        String serviceName = service.getClass().getName();
        Class<?>[] interfaces = service.getClass().getInterfaces();
        for(Class clazz : interfaces){
    
            interfaceProvider.put(clazz.getName(),service);
        }
    }
    public Object getService(String interfaceName){
    
        return interfaceProvider.get(interfaceName);
    }
}

In the previous server code, there are Sevicerprovide This HashMap Where it needs to be changed to ServiProvider, such as

public class TestServer {
    
    public static void main(String[] args) {
    
        UserService userService = new UserServiceImpl();
        BlogService blogService = new BlogServiceImpl();
// Map<String, Object> serviceProvide = new HashMap<>();
// serviceProvide.put("com.ganghuan.myRPCVersion2.service.UserService",userService);
// serviceProvide.put("com.ganghuan.myRPCVersion2.service.BlogService",blogService);
        ServiceProvider serviceProvider = new ServiceProvider();
        serviceProvider.provideServiceInterface(userService);
        serviceProvider.provideServiceInterface(blogService);
        
        RPCServer RPCServer = new SimpleRPCRPCServer(serviceProvider);
        RPCServer.start(8899);
    }
}

result

//  Add new test cases to the customer  
BlogService blogService = rpcClientProxy.getProxy(BlogService.class);
Blog blogById = blogService.getBlogById(10000);
System.out.println(" From the server blog by ：" + blogById);

summary ：

In one version , We refactor the code of the server , The code is more concise ,

Add the server implementation of thread pool version , Performance should improve （ Unmeasured ）

And the server can finally provide different services , More functions , No more chicken ribs

this RPC The biggest pain point

1. Conventional BIO Low network transmission performance with thread pool

---