基于Socket实现PRC通信

梦里梦外; 2023-03-12 14:30 2阅读 0赞

#### 前言 ####

写这篇博客的目的主要回顾一下学过的知识，避免遗忘，同时希望伙伴一起来指正一些不足，共同学习。文章是基于Socket来实现一个基本的RPC通信框架，并且实现版本控制。功能不会太复杂，主要是为了疏通思路脉路。

##### 背景环境 #####

在分布式中，我们经常会用到dubbo+zookeeper的框架来实现，由于本篇博客并没有对zookeeper的实现，所以我们将由RPC-Server来统一对API进行管理，还请见谅！

#### 一、RPC-Server创建 ####

![在这里插入图片描述][20200509170243259.png]

##### 1.1 rpc-server-api #####

###### 1.1.1基本创建 ######

实体类的创建，并实现get&set方法。

//实体类
    public class User { 
    
        private String name;
        private int age;
    
        public String getName() { 
            return name;
        }
    
        public void setName(String name) { 
            this.name = name;
        }
    
        public void setAge(int age) { 
            this.age = age;
        }
    
        public int getAge() { 
            return age;
        }
    
        @Override
        public String toString() { 
            return "User{" +
                    "name='" + name + '\'' +
                    ", age=" + age +
                    '}';
        }
    }

**测试接口：**  
这里随便定义两个方法，就不解释方法了，看名字也能猜出来。

public interface IHelloService { 
    
        String sayHello(String content);
        /** * 保存用户 * @param user * @return */
        String saveUser(User user);
    }

**RpcRequest：**

关于Request对象的解释：

> Request对象，又称为请求对象，该对象派生自HTTPResponse类，是ASP中重要的服务器内置对象，它连接着Web服务器和Web客户端程序。该对象用来获取客户端在请求一个页面或者传送一个Form时提供的所有信息，包括能够标识浏览器和用户的HTTP变量、存储在客户端Cookie信息以及附在URL后面的值、查询字符串或页面中Form段HTML控件内的值、Cookie、客户端证书、查询字符串等 。如浏览器和用户的变量，客户端表单中的数据、变量或者客户端的cookie信息等，Request对象对应的类是System、Web、HttpRequest类。

我们这里也简单存储一下请求的基本信息：

public class RpcRequest implements Serializable { 
        //类名
        private String className;
        //方法名
        private String methodName;
        //参数
        private Object[] parameters;
        //版本号
        private String version;
    
        public String getVersion() { 
            return version;
        }
    
        public void setVersion(String version) { 
            this.version = version;
        }
    
        public String getClassName() { 
            return className;
        }
    
        public void setClassName(String className) { 
            this.className = className;
        }
    
        public String getMethodName() { 
            return methodName;
        }
    
        public void setMethodName(String methodName) { 
            this.methodName = methodName;
        }
    
        public Object[] getParameters() { 
            return parameters;
        }
    
        public void setParameters(Object[] parameters) { 
            this.parameters = parameters;
        }
    }

###### 1.1.2 打jar包 ######

到此处，api模块的编写就简单的完成了。我们将此模块进行打包，然后添加到prc-server-provider中。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNDA5MjYw_size_16_color_FFFFFF_t_70]

##### 1.2 rpc-server-provider #####

在这里我们需要对接口的注册和接收服务端的请求处理，

###### 1.2.1 依赖 ######

<dependency>
                <groupId>com.ccc</groupId>
                <artifactId>rpc-server-api</artifactId>
                <version>1.0-SNAPSHOT</version>
            </dependency>
    
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-context</artifactId>
                <version>4.3.13.RELEASE</version>
            </dependency>

###### 1.2.2 定义RpcService注解 ######

@Target(ElementType.TYPE) //修饰范围 //类或接口
    @Retention(RetentionPolicy.RUNTIME)
    @Component //被spring进行扫描
    public @interface RpcService { 
        
        Class<?> value(); //拿到服务的接口
    
        /** * 版本号 */
        String version() default "";
    }

###### 1.2.3 对api接口实现 ######

由于我们对版本进行了控制，所以此处简单写两个实现类并标注上述声明的注解

@RpcService(value = IHelloService.class,version = "v1.0")
    public class HelloServiceImpl implements IHelloService{ 
    
    
        @Override
        public String sayHello(String content) { 
            System.out.println("[v1.0] request in :"+content);
            return "[v1.0]say Hello:"+content;
        }
    
        @Override
        public String saveUser(User user) { 
            System.out.println("request in saveUser :" +user);
            return "[v1.0]SUCCESS";
        }
    }

@RpcService(value = IHelloService.class,version = "v2.0")
    public class HelloServiceImpl2 implements IHelloService{ 
    
    
        @Override
        public String sayHello(String content) { 
            System.out.println("[v2.0] request in :"+content);
            return "[v2.0]say Hello:"+content;
        }
    
        @Override
        public String saveUser(User user) { 
            System.out.println("request in saveUser :" +user);
            return "[v2.0]SUCCESS";
        }
    }

###### 1.2.4 PrcServer编写 ######

> 创建MyRpcServer类，实现ApplicationContextAware，InitializingBean这两个接口，当然这并不一定是必须需要实现这两个接口，主要能实现功能就行。  
> **ApplicationContextAware**：加载Spring配置文件时，如果Spring配置文件中所定义或者注解自动注入的Bean类实现了ApplicationContextAware 接口，那么在加载Spring配置文件时，会自动调用ApplicationContextAware 接口中的**setApplicationContext**方法。  
> **InitializingBean**：InitializingBean接口为bean提供了初始化方法的方式，它只包括**afterPropertiesSet**方法，凡是继承该接口的类，在初始化bean的时候都会执行该方法。  
> 另外实现的通信是基于Socket套接字来实现的，我们这里使用一个缓存线程池来处理每一次接收到的请求，另外请求的业务逻辑我们也交由ProcessorHandler线程来处理。

public class MyRpcServer implements ApplicationContextAware, InitializingBean { 
         //缓存线程池
        ExecutorService executorService = Executors.newCachedThreadPool();
        //存放注解的容器
        private Map<String,Object> handlerMap = new HashMap<>();
        //端口号
        private int port;
    
        public MyRpcServer(int port){ 
            this.port = port;
        }
    
        /** * InitializingBean接口方法： * 初始化客户端。 * @throws Exception */
        @Override
        public void afterPropertiesSet() throws Exception { 
    
            ServerSocket serverSocket = null;
            try { 
            //创建对象并设置端口
                serverSocket = new ServerSocket(port);
                while (true){ 
                    Socket socket = serverSocket.accept(); //此处会阻塞
                    //每一个socket交给一个processorHandler处理
                    executorService.execute(new ProcessorHandler(handlerMap,socket));
                }
            } catch (IOException e) { 
                e.printStackTrace();
            }finally { 
                if(serverSocket !=null){ 
                    try { 
                        serverSocket.close();
                    } catch (IOException e) { 
                        e.printStackTrace();
                    }
                }
            }
    
        }
    
        /** * ApplicationContextAware接口方法 * 将API进行初始化并存放于容器中 * @param applicationContext * @throws BeansException */
        @Override
        public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { 
        //获取指定注解
            Map<String,Object> serviceMap = applicationContext.getBeansWithAnnotation(RpcService.class);
            if(! serviceMap.isEmpty()){ 
            //遍历注解的value
                for(Object serviceBean : serviceMap.values()){ 
                    //拿到注解
                    RpcService rpcService = serviceBean.getClass().getAnnotation((RpcService.class));
                    //获取name
                    String serviceName = rpcService.value().getName();
                    //获取版本号
                    String version = rpcService.version();
                    //添加版本号
                    if(!StringUtils.isEmpty(version)){ 
                    /** * 将serviceName拼接版本号，在此控制了版本号后, * 如果在注册API时候添加了版本号，那么客户端调用接口的时候，就必须传递版本号信息 * 否则无法进行调用。 */
                        serviceName += "-"+version;
                    }
                    //将name作为key class对象作为value存放
                    handlerMap.put(serviceName,serviceBean);
    
                }
            }
        }
    }

###### 1.2.5 ProcessorHandler ######

> 此类实现了Runnable接口，接收的每个请求都单独用一个线程来处理，该类中主要是从socket中读取客户端发送过来的信息处理然后调用指定的方法。

public class ProcessorHandler implements Runnable { 
    
        private Socket socket;
    
        private Map<String, Object> handlerMap;
    
        /** * 构造器 * @param handlerMap 获取注解信息 * @param socket 获取请求信息 */
        public ProcessorHandler(Map<String, Object> handlerMap, Socket socket) { 
            this.socket = socket;
            this.handlerMap = handlerMap;
        }
    
        @Override
        public void run() { 
            ObjectInputStream objectInputStream = null;
            ObjectOutputStream objectOutputStream = null;
            try { 
                // ------------------- InputStream ------------------
                //拿到客户端信息 输入流
                objectInputStream = new ObjectInputStream(socket.getInputStream());
                /** * 进行反序列化，将客户端发送的Request信息读取出来。 * 包括请求哪个类，方法名称，参数 */
                RpcRequest rpcRequest = (RpcRequest) objectInputStream.readObject();
                //将请求进行处理
                Object result = invoke(rpcRequest);
    
                //-------------------- OutputStream --------------------
                //可以用于发送广播消息。目前没有使用
                objectOutputStream = new ObjectOutputStream(socket.getOutputStream());
                objectOutputStream.writeObject(result);
                objectOutputStream.flush();
            } catch (Exception e) { 
                e.printStackTrace();
            } finally { 
                if (objectInputStream != null) { 
                    try { 
                        objectInputStream.close();
                    } catch (IOException e) { 
                        e.printStackTrace();
                    }
                }
                if (objectOutputStream != null) { 
                    try { 
                        objectOutputStream.close();
                    } catch (IOException e) { 
                        e.printStackTrace();
                    }
                }
            }
        }
    
        private Object invoke(RpcRequest rpcRequest) throws ClassNotFoundException, NoSuchMethodException, InvocationTargetException, IllegalAccessException { 
            //得到类名
            String serviceName = rpcRequest.getClassName();
            //获取版本号
            String version = rpcRequest.getVersion();
            //如果版本号不为空，那么将请求数据按照规定拼接
            if (!StringUtils.isEmpty(version)) { 
                serviceName += "-" + version;
            }
            System.out.println(serviceName);
            System.out.println("map:"+handlerMap);
            //反射调用
            //根据key获取类对象
            Object service = handlerMap.get(serviceName);
    
            //根据RpcRequest请求中的serviceName 如果没有找到 抛出异常。
            if (service == null) { 
                throw new RuntimeException("service not found:" + serviceName);
            }
            //获取请求参数数组
            Object[] args = rpcRequest.getParameters(); //获得请求参数
            Method method = null;
            if (args != null) {  //如果有参数 对参数进行处理，如果没有进行调用加载方法。
                //遍历获得每个参数的类型
                Class<?>[] types = new Class[args.length];
                for (int i = 0; i < args.length; i++) { 
                    //得到参数类型
                    types[i] = args[i].getClass();
                }
                //根据请求的类去加载 //HelloServiceImpl
                Class clazz = Class.forName(rpcRequest.getClassName());
               method = clazz.getMethod(rpcRequest.getMethodName(), types);// sayHello saveUser 找到类中的方法
            }else { 
                //根据请求的类去加载 //HelloServiceImpl
                Class clazz = Class.forName(rpcRequest.getClassName());
                // sayHello saveUser 找到类中的方法
                method = clazz.getMethod(rpcRequest.getMethodName());
            }
    
            //反射调用
            return method.invoke(service, args);
        }
    }

###### 1.2.6 注入 ######

> 将代码交由Spring来管理。

@Configuration
    @ComponentScan(basePackages = "com.ccc")
    public class SpringConfig { 
    
        @Bean(name = "MyRpcServer")
        public MyRpcServer MyRpcServer(){ 
            return new MyRpcServer(9527);
        }
    
    }

###### 1.2.7 启动类 ######

public class App { 
        public static void main(String[] args) { 
    
            ApplicationContext context = new AnnotationConfigApplicationContext(SpringConfig.class);
            ((AnnotationConfigApplicationContext) context).start();
    
        }
    }

###### 1.2.8 打包 ######

将rpc-server打jar包，方法和上图一样。 到这里简单的rpcserver也就完成了。下面写个prc-client

#### 二、RPC-Client 创建 ####

创建个maven项目就行

##### 2.1导入依赖 #####

将RPC-Server的依赖导入进来

<dependency>
          <groupId>com.ccc</groupId>
          <artifactId>rpc-server-api</artifactId>
          <version>1.0-SNAPSHOT</version>
        </dependency>

##### 2.2 与Server进行连接 #####

public class RpcNetTransport { 
        private String host;
        private int prot;
    
        public RpcNetTransport(String host, int prot) { 
            this.host = host;
            this.prot = prot;
        }
    
        public Object send(RpcRequest request) { 
            Socket socket = null;
            Object result = null;
            ObjectOutputStream objectOutputStream = null;
            ObjectInputStream inputStream = null;
    
            try { 
                socket = new Socket(host, prot); //建立连接
                /** * 将客户的端的request信息进行输出到Server端 */
                objectOutputStream = new ObjectOutputStream(socket.getOutputStream());
                objectOutputStream.writeObject(request); //序列化
                objectOutputStream.flush();
    
                // ----------------- InputStream ---------------
                //接受服务端发来的广播消息，暂时没用上。
                inputStream = new ObjectInputStream(socket.getInputStream());
    
                result = inputStream.readObject();
    
            } catch (Exception e) { 
                e.printStackTrace();
            } finally { 
                if (inputStream != null) { 
                    try { 
                        inputStream.close();
                    } catch (IOException e) { 
                        e.printStackTrace();
                    }
                }
                if (objectOutputStream != null) { 
                    try { 
                        objectOutputStream.close();
                    } catch (IOException e) { 
                        e.printStackTrace();
                    }
                }
            }
            return result;
        }
    }

##### 2.3 java动态代理来进行server之间交互。 #####

###### 2.3.1 RemoteInvocationHandler ######

> 在这里我使用java的动态代理来实现创建一个RemoteInvocationHandler实现InvocationHandler，并实现**invoke**方法  
> **InvocationHandler**：是proxy代理实例的调用处理程序实现的一个接口，每一个proxy代理实例都有一个关联的调用处理程序；在代理实例调用方法时，方法调用被编码分派到调用处理程序的invoke方法，该方法有三个参数：  
> **proxy**：代理类代理的真实代理对象com.sun.proxy.$Proxy0  
> **method**：我们所要调用某个对象真实的方法的Method对象  
> **args**：指代代理对象方法传递的参数

public class RemoteInvocationHandler implements InvocationHandler { 
    
        private String host;
        private int prot;
    
        public RemoteInvocationHandler(String host, int prot) { 
            this.host = host;
            this.prot = prot;
        }
    
        @Override
        public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { 
            //打桩
            System.out.println("come in");
            RpcRequest rpcRequest = new RpcRequest();
            // --------------- 开始 设置 RpcRequest 请求参数 ------------------
            //设置类名
            rpcRequest.setClassName(method.getDeclaringClass().getName());
            //设置方法名
            rpcRequest.setMethodName(method.getName());
            //参数
            rpcRequest.setParameters(args);
            //版本号
            rpcRequest.setVersion("v1.0");
            // --------------- 结束 设置 RpcRequest 请求参数 ------------------
    
            //远程通信,将rpcRequest进行输出。
            RpcNetTransport netTransport = new RpcNetTransport(host,prot);
            Object result = netTransport.send(rpcRequest);
            //返回结果
            return result;
        }
    }

###### 2.3.2 RpcClient ######

> 完成了上述操作，现在我就只需要对接口进行代理处理来实现RPC通信了。

public class RpcProxyClient { 
    
        public <T> T clientProxy(final Class<T> interfaceCls,final String host,final int port){ 
            /** * 使用java自带的代理类 将接口进行代理来进行 RPC远程通信，该方法会自动去调用 * InvocationHandler 中的invok方法 */
           return (T) Proxy.newProxyInstance(interfaceCls.getClassLoader(),
                    new Class<?>[]{ interfaceCls},new RemoteInvocationHandler(host,port));
        }
    }

###### 2.3.3 启动类 ######

> 我们只需要写个main方法来启动就行

public class App 
    { 
        public static void main( String[] args )
        { 
            System.out.println( "Hello World!" );
            RpcProxyClient rpcProxyClient = new RpcProxyClient();
            //传入需要调用的接口,主机地址，端口号
            IHelloService iHelloService = rpcProxyClient.clientProxy(IHelloService.class,"localhost",9527);
            //接口调用
            String result = iHelloService.sayHello("Ccc");
            //输出返回结果
            System.out.println(result);
        }
    }

#### 三、测试。 ####

启动Rpc-Server…没有报错  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNDA5MjYw_size_16_color_FFFFFF_t_70 1]  
启动Rpc-Client…  
我们查看一下server接收到的参数  
发现客户端的标识是 Ccc  
![在这里插入图片描述][20200509165100475.png]  
客户端接收的返回结果

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNDA5MjYw_size_16_color_FFFFFF_t_70 2]  
接口调用成功。。

**我们测试一下 不添加版本号调用：**  
在RemoteInvocationHandler 中取出对version的设置。。

**客户端报错：**  
原因很简单，因为我们在Server端对接口的的注解@RpcService中进行了版本控制所以serverName应该是serverName-version的格式，而且我们是通过serverName作为key来获取类对象的。而在客户端不传递的version版本的时候，我们默认是使用serverName的，所以服务端通过这个key就没法找到对应的value。  
![在这里插入图片描述][20200509165241598.png]  
**我们在修改version为v2.0**  
![在这里插入图片描述][20200509165541366.png]  
version2.0 输出成功。

到此，简单的通信功能也就完成，如果您发现有啥错误的地方，恳请指出批评。在此感谢！

[20200509170243259.png]: /images/20230312/50997f44703740aa8bb3536a6dfc33aa.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNDA5MjYw_size_16_color_FFFFFF_t_70]: /images/20230312/d50f45c270054425879928b7678306f9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNDA5MjYw_size_16_color_FFFFFF_t_70 1]: /images/20230312/f757ad706be4414a844830f3620085ce.png
[20200509165100475.png]: /images/20230312/07235b9ee13045a1ad5b12418bc463b4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNDA5MjYw_size_16_color_FFFFFF_t_70 2]: /images/20230312/5884416e8a6745f0a5e0b554823fc0f9.png
[20200509165241598.png]: /images/20230312/aaab961000f443daafcee5ad9fbd9575.png
[20200509165541366.png]: /images/20230312/4deb4687b47a43eda7777ba153ea92ae.png