关于rpc:java-从零开始手写-RPC-01-基于-websocket-实现

27次阅读

共计 3642 个字符,预计需要花费 10 分钟才能阅读完成。

RPC

解决的问题

RPC 次要是为了解决的两个问题:

  • 解决分布式系统中,服务之间的调用问题。
  • 近程调用时,要可能像本地调用一样不便,让调用者感知不到近程调用的逻辑。

这一节咱们来学习下如何基于 websocket 实现最简略的 rpc 调用,后续会实现基于 netty4 的版本。

开源地址:https://github.com/houbb/rpc

残缺流程

其中右边的 Client,对应的就是后面的 Service A,而左边的 Server,对应的则是 Service B。

上面一步一步具体解释一下。

  1. Service A 的应用层代码中,调用了 Calculator 的一个实现类的 add 办法,心愿执行一个加法运算;
  2. 这个 Calculator 实现类,外部并不是间接实现计算器的加减乘除逻辑,而是通过近程调用 Service B 的 RPC 接口,来获取运算后果,因而称之为 Stub;
  3. Stub 怎么和 Service B 建设近程通信呢?这时候就要用到近程通信工具了,也就是图中的 Run-time Library,这个工具将帮你实现近程通信的性能,比方 Java 的 Socket,就是这样一个库,当然,你也能够用基于 Http 协定的 HttpClient,或者其余通信工具类,都能够,RPC 并没有规定说你要用何种协定进行通信;
  4. Stub 通过调用通信工具提供的办法,和 Service B 建设起了通信,而后将申请数据发给 Service B。须要留神的是,因为底层的网络通讯是基于二进制格局的,因而这里 Stub 传给通信工具类的数据也必须是二进制,比方 calculator.add(1,2),你必须把参数值 1 和 2 放到一个 Request 对象外头(这个 Request 对象当然不只这些信息,还包含要调用哪个服务的哪个 RPC 接口等其余信息),而后序列化为二进制,再传给通信工具类,这一点也将在上面的代码实现中体现;
  5. 二进制的数据传到 Service B 这一边了,Service B 当然也有本人的通信工具,通过这个通信工具接管二进制的申请;
  6. 既然数据是二进制的,那么天然要进行反序列化了,将二进制的数据反序列化为申请对象,而后将这个申请对象交给 Service B 的 Stub 解决;
  7. 和之前的 Service A 的 Stub 一样,这里的 Stub 也同样是个“假玩意”,它所负责的,只是去解析申请对象,晓得调用方要调的是哪个 RPC 接口,传进来的参数又是什么,而后再把这些参数传给对应的 RPC 接口,也就是 Calculator 的理论实现类去执行。很显著,如果是 Java,那这里必定用到了反射。
  8. RPC 接口执行结束,返回执行后果,当初轮到 Service B 要把数据发给 Service A 了,怎么发?一样的情理,一样的流程,只是当初 Service B 变成了 Client,Service A 变成了 Server 而已:Service B 反序列化执行后果 -> 传输给 Service A->Service A 反序列化执行后果 -> 将后果返回给 Application,结束。

简略实现

假如服务 A,想调用服务 B 的一个办法。

因为不在同一个内存中,无奈间接应用。如何能够实现相似 Dubbo 的性能呢?

这里不须要应用 HTTP 级别的通信,应用 TCP 协定即可。

common

专用模块,定义通用对象。

  • Rpc 常量
public interface RpcConstant {

    /**
     * 地址
     */
    String ADDRESS = "127.0.0.1";

    /**
     * 端口号
     */
    int PORT = 12345;

}
  • 申请入参
public class RpcCalculateRequest implements Serializable {

    private static final long serialVersionUID = 6420751004355300996L;

    /**
     * 参数一
     */
    private int one;

    /**
     * 参数二
     */
    private int two;

    //getter & setter & toString()}
  • 服务接口
public interface Calculator {

    /**
     * 计算加法
     * @param one 参数一
     * @param two 参数二
     * @return 返回后果
     */
    int add(int one, int two);

}

server

  • 服务接口的实现
public class CalculatorImpl implements Calculator {

    @Override
    public int add(int one, int two) {return one + two;}

}
  • 启动服务
public static void main(String[] args) throws IOException {Calculator calculator = new CalculatorImpl();
    try (ServerSocket listener = new ServerSocket(RpcConstant.PORT)) {System.out.println("Server 端启动:" + RpcConstant.ADDRESS + ":" + RpcConstant.PORT);
        while (true) {try (Socket socket = listener.accept()) {
                // 将申请反序列化
                ObjectInputStream objectInputStream = new ObjectInputStream(socket.getInputStream());
                Object object = objectInputStream.readObject();
                System.out.println("Request is:" + object);
                // 调用服务
                int result = 0;
                if (object instanceof RpcCalculateRequest) {RpcCalculateRequest calculateRpcRequest = (RpcCalculateRequest) object;
                    result = calculator.add(calculateRpcRequest.getOne(), calculateRpcRequest.getTwo());
                }
                // 返回后果
                ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream());
                objectOutputStream.writeObject(result);
            } catch (Exception e) {e.printStackTrace();
            }
        }
    }
}

启动日志:

Server 端启动:127.0.0.1:12345

client

  • 客户端调用
public static void main(String[] args) {Calculator calculator = new CalculatorProxy();
    int result = calculator.add(1, 2);
    System.out.println(result);
}
  • 计算的代理类
public class CalculatorProxy implements Calculator {

    @Override
    public int add(int one, int two) {
        try {Socket socket = new Socket(RpcConstant.ADDRESS, RpcConstant.PORT);

            // 将申请序列化
            RpcCalculateRequest calculateRpcRequest = new RpcCalculateRequest(one, two);
            ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream());

            // 将申请发给服务提供方
            objectOutputStream.writeObject(calculateRpcRequest);

            // 将响应体反序列化
            ObjectInputStream objectInputStream = new ObjectInputStream(socket.getInputStream());
            Object response = objectInputStream.readObject();

            if (response instanceof Integer) {return (Integer) response;
            } else {throw new RuntimeException();
            }
        } catch (IOException | ClassNotFoundException e) {throw new RuntimeException(e);
        }
    }
}
  • 调用日志

client 端

3

server 端

Server 端启动:127.0.0.1:12345
Request is: RpcCalculateRequest{one=1, two=2}

开源地址

为了便于大家学习,以上源码曾经开源:

https://github.com/houbb/rpc

我是老马,期待与你的下次重逢。

正文完
 0