JAVA RPC（一）RPC入門

為什么要寫這個RPC

      市面上常見的RPC框架很多，grpc，motan，dubbo等，但是隨着越來越多的元素加入，復雜的架構設計等因素似使得這些框架就想spring一樣，雖然號稱是輕量級，但是用起來卻是讓我們很蹩腳，大量的配置，繁雜的API設計，其實，我們根本用不上這些東西！！！ 我也算得上是在很多個互聯網企業廝殺過，見過很多很多的內部RPC框架，有些優秀的設計讓我非常贊賞，有一天我突然想着，為什么不對這些設計原型進行聚合歸類，自己搞一套【輕量級】RPC框架呢，礙於工作原因，一直沒有時間倒騰出空，十一期間工作閑暇，說搞就搞吧，希望源碼對大家對認識RPC框架起到推進的作用。東西越寫越多，有各種問題歡迎隨時拍磚

RPC基礎概念

     RPC就是遠程過程調用協議，其作用就是客戶端與服務端之間的遠程調用，就像本地自己調用一樣，讓服務端進行服務化，功能唯一性，負載熱點流量。

RPC框架的實現

    RPC能夠讓本地應用簡單、高效地調用服務器中的過程（服務）。它主要應用在分布式系統。如Hadoop中的IPC組件。但怎樣實現一個RPC框架呢？

從下面幾個方面思考，僅供參考：

1.通信模型：假設通信的為A機器與B機器，A與B之間有通信模型，在Java中一般基於BIO或NIO；。

2.過程（服務）定位：使用給定的通信方式，與確定IP與端口及方法名稱確定具體的過程或方法；

3.遠程代理對象：本地調用的方法(服務)其實是遠程方法的本地代理，因此可能需要一個遠程代理對象，對於Java而言，遠程代理對象可以使用Java的動態對象實現，封裝了調用遠程方法調用；

4.序列化，將對象名稱、方法名稱、參數等對象信息進行網絡傳輸需要轉換成二進制傳輸，這里可能需要不同的序列化技術方案。如:thrift,protobuf,Arvo等。

 

下面用簡單的原生java socket來實現rpc調用，方便大家更深層了解rpc原理

服務類接口

 

package socketrpc;

public interface IHello {
     String sayHello(String string);
}

 

服務實現類

package socketrpc.server;

import socketrpc.IHello;

public class HelloServiceImpl implements IHello {

    @Override
    public String sayHello(String string) {
        return "你好:".concat ( string );
    }
}

 

客戶端實現

package socketrpc.client;

import socketrpc.IHello;

import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.net.InetSocketAddress;
import java.net.Socket;

public class RpcClientProxy<T> implements InvocationHandler {

    private  Class<T> serviceInterface;
    private InetSocketAddress addr;

    public RpcClientProxy(Class<T> serviceInterface, String ip,String port) {
        this.serviceInterface = serviceInterface;
        this.addr = new InetSocketAddress(ip, Integer.parseInt ( port ));
    }

    public T getClientIntance(){
        return (T) Proxy.newProxyInstance (serviceInterface.getClassLoader(),new Class<?>[]{serviceInterface},this);
    }

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {

        Socket socket = null;
        ObjectOutputStream output = null;
        ObjectInputStream input = null;

        try {
            // 2.創建Socket客戶端，根據指定地址連接遠程服務提供者
            socket = new Socket();
            socket.connect(addr);

            // 3.將遠程服務調用所需的接口類、方法名、參數列表等編碼后發送給服務提供者
            output = new ObjectOutputStream(socket.getOutputStream());
            output.writeUTF(serviceInterface.getName());
            output.writeUTF(method.getName());
            output.writeObject(method.getParameterTypes());
            output.writeObject(args);

            // 4.同步阻塞等待服務器返回應答，獲取應答后返回
            input = new ObjectInputStream(socket.getInputStream());
            return input.readObject();
        } finally {
            if (socket != null) socket.close();
            if (output != null) output.close();
            if (input != null) input.close();
        }
    }

    public static void main(String[] args) {
        RpcClientProxy client = new RpcClientProxy<>(IHello.class,"localhost","6666");
        IHello hello = (IHello) client.getClientIntance ();
        System.out.println (hello.sayHello ( "socket rpc" ));
    }
}

 

服務端實現

package socketrpc.server;

import socketrpc.IHello;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.HashMap;

public class RpcServer {

    private static final HashMap<String, Class<?>> serviceRegistry = new HashMap<>();
    private  int port;

    public RpcServer(int port) {
        this.port =port;
    }

    public RpcServer register(Class serviceInterface, Class impl) {
        serviceRegistry.put(serviceInterface.getName(), impl);
        return this;
    }

    public void run() throws IOException {

        ServerSocket server = new ServerSocket();
        server.bind(new InetSocketAddress (port));
        System.out.println("start server");
        ObjectInputStream input =null;
        ObjectOutputStream output =null;
        Socket socket=null;
        try {
            while(true){
                socket = server.accept ();
                input =new ObjectInputStream(socket.getInputStream());
                String serviceName = input.readUTF();
                String methodName = input.readUTF();
                System.out.println (methodName);
                Class<?>[] parameterTypes = (Class<?>[]) input.readObject();
                Object[] arguments = (Object[]) input.readObject();
                Class serviceClass = serviceRegistry.get(serviceName);
                if (serviceClass == null) {
                    throw new ClassNotFoundException(serviceName + " not found");
                }
                Method method = serviceClass.getMethod(methodName, parameterTypes);
                Object result = method.invoke(serviceClass.newInstance(), arguments);
                output = new ObjectOutputStream (socket.getOutputStream());
                output.writeObject(result);
            }
        } catch (Exception e){
            e.printStackTrace();
        }finally {
            if (output != null) {
                try {
                    output.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            if (input != null) {
                try {
                    input.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            if (socket != null) {
                try {
                    socket.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

    }

    public static void main(String[] args) throws IOException {
        new RpcServer ( 6666 ).register ( IHello.class,HelloServiceImpl.class).run ();
    }

}

啟動服務端和客戶端后運行結果如下

 

 

 

 

客戶端代碼簡單講解

 private  Class<T> serviceInterface;
    private InetSocketAddress addr;

    public RpcClientProxy(Class<T> serviceInterface, String ip,String port) {
        this.serviceInterface = serviceInterface;
        this.addr = new InetSocketAddress(ip, Integer.parseInt ( port ));
    }

    public T getClientIntance(){
        return (T) Proxy.newProxyInstance (serviceInterface.getClassLoader(),new Class<?>[]{serviceInterface},this);
    }

傳入接口類和ip端口,調用getClientIntance方法時,對當前接口進行代理,實際調用方法為

@Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {

        Socket socket = null;
        ObjectOutputStream output = null;
        ObjectInputStream input = null;

        try {
            // 2.創建Socket客戶端，根據指定地址連接遠程服務提供者
            socket = new Socket();
            socket.connect(addr);

            // 3.將遠程服務調用所需的接口類、方法名、參數列表等編碼后發送給服務提供者
            output = new ObjectOutputStream(socket.getOutputStream());
            output.writeUTF(serviceInterface.getName());
            output.writeUTF(method.getName());
            output.writeObject(method.getParameterTypes());
            output.writeObject(args);

            // 4.同步阻塞等待服務器返回應答，獲取應答后返回
            input = new ObjectInputStream(socket.getInputStream());
            return input.readObject();
        } finally {
            if (socket != null) socket.close();
            if (output != null) output.close();
            if (input != null) input.close();
        }
    }

遵循下面步驟

1：創建socket，並和遠程進行三次連接握手【socket.connect(addr)】。

2：封裝socket輸出流【ObjectOutputStream】。

3：輸出類名稱，方法名稱，參數類型和參數值給server。

4：獲取socket輸入流，等待server返回結果。

 

服務端代碼簡單講解

public RpcServer(int port) {
        this.port =port;
    }

    public RpcServer register(Class serviceInterface, Class impl) {
        serviceRegistry.put(serviceInterface.getName(), impl);
        return this;
    }

初始化服務端，將服務類注冊到hashMap【模擬spring上下文】

public void run() throws IOException {

        ServerSocket server = new ServerSocket();
        server.bind(new InetSocketAddress (port));
        System.out.println("start server");
        ObjectInputStream input =null;
        ObjectOutputStream output =null;
        Socket socket=null;
        try {
            while(true){
                socket = server.accept ();
                input =new ObjectInputStream(socket.getInputStream());

                String serviceName = input.readUTF();
                String methodName = input.readUTF();
                System.out.println (methodName);
                Class<?>[] parameterTypes = (Class<?>[]) input.readObject();
                Object[] arguments = (Object[]) input.readObject();
                Class serviceClass = serviceRegistry.get(serviceName);
                if (serviceClass == null) {
                    throw new ClassNotFoundException(serviceName + " not found");
                }
                Method method = serviceClass.getMethod(methodName, parameterTypes);
                Object result = method.invoke(serviceClass.newInstance(), arguments);
                output = new ObjectOutputStream (socket.getOutputStream());
                output.writeObject(result);
            }
        } catch (Exception e){
            e.printStackTrace();
        }finally {
            if (output != null) {
                try {
                    output.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            if (input != null) {
                try {
                    input.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            if (socket != null) {
                try {
                    socket.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

    }

服務端執行做了以下幾件事：

1：綁定端口，阻塞等待客戶端調用【socket = server.accept ()】。

2：封裝輸入流【socket.getInputStream()】。

3：從輸入流中獲取到接口名，方法名，參數類型，參數值。

4：找到初始化時hashmap中的服務類。

5：反射獲取服務實現類方法並根據請求參數進行服務調用。

6：封裝輸出流【ObjectOutputStream】，並且返回結果。

 

到目前為止，整個簡單的socket實現的RPC服務就已經全部完成了，可以優化的部分。

1：序列化局限，原生序列化只能序列化實現了【Serializable】接口的服務類，並且序列化復雜對象時，內容龐大效率極低，需要高效的序列化協議進行序列化參數方法等必要請求入參

2：BIO性能局限，socket服務端采用默認的BIO來阻塞獲取輸入流，效率低下，需采用NIO等異步非阻塞服務端方案，例如netty,mina和java nio等。

3：在大型企業級RPC解決方案中，客戶端和服務端的長連接需要一直保持，否則每次調用時都要重新進行三次握手和四次揮手，這樣頻繁的創建tcp連接對機器性能是極大的損耗，對socket的連接可以采用apache pool2連接池等方案

4：服務端負載，需要考慮服務自動發現，讓客戶端在不需要重啟的情況下能動態感知服務端的變化，從而實現熱部署等。可以采用辦法定時自動輪詢，zookeeper等。

5：服務端服務類執行異常，客戶端感知等。

 

好了，相信看完本章內容對於rpc框架來說，大家已經將基礎了解的差不多了，下面我將會給大家全面講解基於zk，thrift，netty的企業級RPC解決方案