文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

Java 从零开始手写 RPC基于 Websocket 实现

2024-12-02 20:01

关注

(1)解决分布式系统中,服务之间的调用问题。

(2)远程调用时,要能够像本地调用一样方便,让调用者感知不到远程调用的逻辑。

这一节我们来学习下如何基于 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

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

  1. public interface RpcConstant { 
  2.  
  3.  
  4.      
  5.     String ADDRESS = "127.0.0.1"
  6.  
  7.  
  8.      
  9.     int PORT = 12345; 
  10.  
  11.  
  1. public class RpcCalculateRequest implements Serializable { 
  2.  
  3.  
  4.     private static final long serialVersionUID = 6420751004355300996L; 
  5.  
  6.  
  7.      
  8.     private int one; 
  9.  
  10.  
  11.      
  12.     private int two; 
  13.  
  14.  
  15.     //getter & setter & toString() 
  1. public interface Calculator { 
  2.  
  3.  
  4.      
  5.     int add(int one, int two); 
  6.  
  7.  

server

  1. public class CalculatorImpl implements Calculator { 
  2.  
  3.  
  4.     @Override 
  5.     public int add(int one, int two) { 
  6.         return one + two; 
  7.     } 
  8.  
  9.  
  1. public static void main(String[] args) throws IOException { 
  2.     Calculator calculator = new CalculatorImpl(); 
  3.     try (ServerSocket listener = new ServerSocket(RpcConstant.PORT)) { 
  4.         System.out.println("Server 端启动:" + RpcConstant.ADDRESS + ":" + RpcConstant.PORT); 
  5.         while (true) { 
  6.             try (Socket socket = listener.accept()) { 
  7.                 // 将请求反序列化 
  8.                 ObjectInputStream objectInputStream = new ObjectInputStream(socket.getInputStream()); 
  9.                 Object object = objectInputStream.readObject(); 
  10.                 System.out.println("Request is: " + object); 
  11.                 // 调用服务 
  12.                 int result = 0; 
  13.                 if (object instanceof RpcCalculateRequest) { 
  14.                     RpcCalculateRequest calculateRpcRequest = (RpcCalculateRequest) object; 
  15.                     result = calculator.add(calculateRpcRequest.getOne(), calculateRpcRequest.getTwo()); 
  16.                 } 
  17.                 // 返回结果 
  18.                 ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream()); 
  19.                 objectOutputStream.writeObject(result); 
  20.             } catch (Exception e) { 
  21.                 e.printStackTrace(); 
  22.             } 
  23.         } 
  24.     } 

启动日志:

  1. Server 端启动:127.0.0.1:12345 

client

•客户端调用

  1. public static void main(String[] args) { 
  2.     Calculator calculator = new CalculatorProxy(); 
  3.     int result = calculator.add(1, 2); 
  4.     System.out.println(result); 
  1. public class CalculatorProxy implements Calculator { 
  2.  
  3.  
  4.     @Override 
  5.     public int add(int one, int two) { 
  6.         try { 
  7.             Socket socket = new Socket(RpcConstant.ADDRESS, RpcConstant.PORT); 
  8.  
  9.  
  10.             // 将请求序列化 
  11.             RpcCalculateRequest calculateRpcRequest = new RpcCalculateRequest(one, two); 
  12.             ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream()); 
  13.  
  14.  
  15.             // 将请求发给服务提供方 
  16.             objectOutputStream.writeObject(calculateRpcRequest); 
  17.  
  18.  
  19.             // 将响应体反序列化 
  20.             ObjectInputStream objectInputStream = new ObjectInputStream(socket.getInputStream()); 
  21.             Object response = objectInputStream.readObject(); 
  22.  
  23.  
  24.             if (response instanceof Integer) { 
  25.                 return (Integer) response; 
  26.             } else { 
  27.                 throw new RuntimeException(); 
  28.             } 
  29.         } catch (IOException | ClassNotFoundException e) { 
  30.             throw new RuntimeException(e); 
  31.         } 
  32.     } 

client 端

server 端

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

 

来源:今日头条内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     813人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     435人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-后端开发
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯