导航：首页 > 互联网科技 >

java rpc框架的原理是什么

发表于：2024-11-26 作者：千家信息网编辑

千家信息网最后更新 2024年11月26日，本篇内容主要讲解"java rpc框架的原理是什么"，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习"java rpc框架的原理是什么"吧!RPC基础概念RPC

千家信息网最后更新 2024年11月26日java rpc框架的原理是什么

本篇内容主要讲解"java rpc框架的原理是什么"，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习"java 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;2 3 public interface IHello {4      String sayHello(String string);5 }

服务实现类

1 package socketrpc.server; 2  3 import socketrpc.IHello; 4  5 public class HelloServiceImpl implements IHello { 6  7     @Override 8     public String sayHello(String string) { 9         return "你好:".concat ( string );10     }11 }

客户端实现

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 implements InvocationHandler {    private  Class serviceInterface;    private InetSocketAddress addr;    public RpcClientProxy(Class 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> 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 serviceInterface; 2     private InetSocketAddress addr; 3  4     public RpcClientProxy(Class serviceInterface, String ip,String port) { 5         this.serviceInterface = serviceInterface; 6         this.addr = new InetSocketAddress(ip, Integer.parseInt ( port )); 7     } 8  9     public T getClientIntance(){10         return (T) Proxy.newProxyInstance (serviceInterface.getClassLoader(),new Class[]{serviceInterface},this);11     }

传入接口类和ip端口,调用getClientIntance方法时,对当前接口进行代理,实际调用方法为

1 @Override 2     public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { 3  4         Socket socket = null; 5         ObjectOutputStream output = null; 6         ObjectInputStream input = null; 7  8         try { 9             // 2.创建Socket客户端，根据指定地址连接远程服务提供者10             socket = new Socket();11             socket.connect(addr);12 13             // 3.将远程服务调用所需的接口类、方法名、参数列表等编码后发送给服务提供者14             output = new ObjectOutputStream(socket.getOutputStream());15             output.writeUTF(serviceInterface.getName());16             output.writeUTF(method.getName());17             output.writeObject(method.getParameterTypes());18             output.writeObject(args);19 20             // 4.同步阻塞等待服务器返回应答，获取应答后返回21             input = new ObjectInputStream(socket.getInputStream());22             return input.readObject();23         } finally {24             if (socket != null) socket.close();25             if (output != null) output.close();26             if (input != null) input.close();27         }28     }

遵循下面步骤

1：创建socket，并和远程进行三次连接握手【socket.connect(addr)】。

2：封装socket输出流【ObjectOutputStream】。

3：输出类名称，方法名称，参数类型和参数值给server。

4：获取socket输入流，等待server返回结果。

服务端代码简单讲解

1 public RpcServer(int port) {2         this.port =port;3     }4 5     public RpcServer register(Class serviceInterface, Class impl) {6         serviceRegistry.put(serviceInterface.getName(), impl);7         return this;8     }

初始化服务端，将服务类注册到hashMap【模拟spring上下文】

1 public void run() throws IOException { 2  3         ServerSocket server = new ServerSocket(); 4         server.bind(new InetSocketAddress (port)); 5         System.out.println("start server"); 6         ObjectInputStream input =null; 7         ObjectOutputStream output =null; 8         Socket socket=null; 9         try {10             while(true){11                 socket = server.accept ();12                 input =new ObjectInputStream(socket.getInputStream());13 14                 String serviceName = input.readUTF();15                 String methodName = input.readUTF();16                 System.out.println (methodName);17                 Class[] parameterTypes = (Class[]) input.readObject();18                 Object[] arguments = (Object[]) input.readObject();19                 Class serviceClass = serviceRegistry.get(serviceName);20                 if (serviceClass == null) {21                     throw new ClassNotFoundException(serviceName + " not found");22                 }23                 Method method = serviceClass.getMethod(methodName, parameterTypes);24                 Object result = method.invoke(serviceClass.newInstance(), arguments);25                 output = new ObjectOutputStream (socket.getOutputStream());26                 output.writeObject(result);27             }28         } catch (Exception e){29             e.printStackTrace();30         }finally {31             if (output != null) {32                 try {33                     output.close();34                 } catch (IOException e) {35                     e.printStackTrace();36                 }37             }38             if (input != null) {39                 try {40                     input.close();41                 } catch (IOException e) {42                     e.printStackTrace();43                 }44             }45             if (socket != null) {46                 try {47                     socket.close();48                 } catch (IOException e) {49                     e.printStackTrace();50                 }51             }52         }53 54     }

服务端执行做了以下几件事：

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等。