Myrpc version 5

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

---

5.MyRPC edition 5

Background knowledge

• zookeeper install , Basic concepts
• understand curator Open source zookeeper Use in the client

Questions in this section

• How to design a registry

Registry Center （ Such as zookeeper） The address is fixed （ For high availability, it is usually a cluster , We can regard it as a black box ）, When the server goes online , Register your service and corresponding address in the registry , When the client invokes the service , Go to the registry and find the corresponding server address according to the service name .

zookeeper We can approximate it as a tree directory file system , Is a distributed coordination application , Other registries are EureKa, Nacos etc.

Upgrade process

Premise

1. Download decompression Zookeeper [ Address ]（https://zookeeper.apache.org/releases.html）
2. Learn one zookeeper Examples of startup The official example
   1. zoo.cfg modify dataDir For an existing directory
   2. windows Start command : bin/zkServer.cmd
3. java Introduced in the project Curator client ,

<!-- This jar Packages should depend on log4j, Do not introduce log4j There will be consoles there will be warn, But it does not affect normal use -->
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-recipes</artifactId>
    <version>5.1.0</version>
</dependency>

to update 1 ： introduce zookeeper As a registry

Start a local zookeeper Server side , Default port 2181.zookeeper Client tests are as follows ：

First define the service registration interface

//  Service registration interface , Two basic functions , register ： Save service and address .  Inquire about ： Find the address according to the service name 
public interface ServiceRegister {
    
    void register(String serviceName, InetSocketAddress serverAddress);
    InetSocketAddress serviceDiscovery(String serviceName);
}

zookeeper Implementation class of service registration interface

public class ZkServiceRegister implements ServiceRegister{
    
    // curator  Provided zookeeper client 
    private CuratorFramework client;
    // zookeeper Root path node 
    private static final String ROOT_PATH = "MyRPC";
    //  This is responsible zookeeper Initialization of the client , And with zookeeper The server establishes a connection 
    public ZkServiceRegister(){
    
        //  Exponential time retry 
        RetryPolicy policy = new ExponentialBackoffRetry(1000, 3);
        // zookeeper Your address is fixed , Whether it's a service provider or , Consumers have to connect with it 
        // sessionTimeoutMs  And  zoo.cfg Medium tickTime  It matters ,
        // zk And according to minSessionTimeout And maxSessionTimeout Two parameters readjust the last timeout value . The default values are tickTime  Of 2 Times and 20 times 
        //  Use heartbeat monitoring status 
        this.client = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181")
                .sessionTimeoutMs(40000).retryPolicy(policy).namespace(ROOT_PATH).build();
        this.client.start();
        System.out.println("zookeeper  Successful connection ");
    }
    @Override
    public void register(String serviceName, InetSocketAddress serverAddress){
    
        try {
    
            // serviceName Create a permanent node , When the service provider goes offline , Do not delete the service name , Just delete the address 
            if(client.checkExists().forPath("/" + serviceName) == null){
    
               client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/" + serviceName); 
            }
            //  Path address , One / Represents a node 
            String path = "/" + serviceName +"/"+ getServiceAddress(serverAddress);
            //  Temporary node , Delete the node when the server goes offline 
            client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(path);
        } catch (Exception e) {
    
            System.out.println(" This service already exists ");
        }
    }
    //  Return the address according to the service name 
    @Override
    public InetSocketAddress serviceDiscovery(String serviceName) {
    
        try {
    
            List<String> strings = client.getChildren().forPath("/" + serviceName);
            //  The first one used here by default , Then add load balancing 
            String string = strings.get(0);
            return parseAddress(string);
        } catch (Exception e) {
    
            e.printStackTrace();
        }
        return null;
    }
    //  Address  -> XXX.XXX.XXX.XXX:port  character string 
    private String getServiceAddress(InetSocketAddress serverAddress) {
    
        return serverAddress.getHostName() +
                ":" +
                serverAddress.getPort();
    }
    //  The string resolves to an address 
    private InetSocketAddress parseAddress(String address) {
    
        String[] result = address.split(":");
        return new InetSocketAddress(result[0], Integer.parseInt(result[1]));
    }
}

to update 2： Update how the client gets the server , When the server exposes Services , Register with the registry

First new client There is no need to pass in host And name, While sending request when , From the registry

//  No transmission required host,port
RPCClient rpcClient = new NettyRPCClient();

Client transformation

public class SimpleRPCClient implements RPCClient {
    
    private String host;
    private int port;
    private ServiceRegister serviceRegister;
    public SimpleRPCClient() {
    
        //  Initialize registry , Establishing a connection 
        this.serviceRegister = new ZkServiceRegister();
    }
    //  The client initiates a request call ,Socket Establishing a connection , Initiate request Request, Get a response Response
    //  there request It's packaged , Different service Different packaging is required ,  The client only knows Service Interface , A layer of dynamic proxy is needed to encapsulate different objects according to reflection Service
    public RPCResponse sendRequest(RPCRequest request) {
    
        //  Get... From the registry host,port
        InetSocketAddress address = serviceRegister.serviceDiscovery(request.getInterfaceName());
        host = address.getHostName();
        port = address.getPort();
        
        try {
    
            Socket socket = new Socket(host, port);
            ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream());
            ObjectInputStream objectInputStream = new ObjectInputStream(socket.getInputStream());
            System.out.println(request);
            objectOutputStream.writeObject(request);
            objectOutputStream.flush();
            RPCResponse response = (RPCResponse) objectInputStream.readObject();
            //System.out.println(response.getData());
            return response;
        } catch (IOException | ClassNotFoundException e) {
    
            System.out.println();
            return null;
        }
    }
}

The transformation of the server ： Instead, the server needs to put its own ip, Port to registry

ServiceProvider serviceProvider = new ServiceProvider("127.0.0.1", 8899);

Add the function of registration in the service exposure class

public class ServiceProvider {
    
    /** *  An implementation class may implement multiple service interfaces , */
    private Map<String, Object> interfaceProvider;
    private ServiceRegister serviceRegister;
    private String host;
    private int port;
    public ServiceProvider(String host, int port){
    
        //  You need to pass in the network address of the server's own service 
        this.host = host;
        this.port = port;
        this.interfaceProvider = new HashMap<>();
        this.serviceRegister = new ZkServiceRegister();
    }
    public void provideServiceInterface(Object service){
    
        Class<?>[] interfaces = service.getClass().getInterfaces();
        for(Class clazz : interfaces){
    
            //  Native mapping table 
            interfaceProvider.put(clazz.getName(),service);
            //  Register the service in the registry 
            serviceRegister.register(clazz.getName(),new InetSocketAddress(host,port));
        }
    }
    public Object getService(String interfaceName){
    
        return interfaceProvider.get(interfaceName);
    }
}

result

Running successfully ！

summary

In this version, we have added the registry , Finally a complete RPC The framework has three roles ： Service providers , Serving consumers , Registry Center

The biggest pain point of this version

• When querying the address according to the service name , We always return first IP, This leads to great pressure on the provider , Other providers cannot call

---