RPC框架简易实现

RPC（Remote Procedure Call Protocol）——远程过程调用协议，它是一种通过网络从远程计算机程序上请求服务，而不需要了解底层网络技术的协议。RPC协议假定某些传输协议的存在，如TCP或UDP，为通信程序之间携带信息数据。在OSI网络通信模型中，RPC跨越了传输层和应用层。RPC使得开发包括网络分布式多程序在内的应用程序更加容易。

RPC采用客户机/服务器模式。请求程序就是一个客户机，而服务提供程序就是一个服务器。首先，客户机调用进程发送一个有进程参数的调用信息到服务进程，然后等待应答信息。在服务器端，进程保持睡眠状态直到调用信息到达为止。当一个调用信息到达，服务器获得进程参数，计算结果，发送答复信息，然后等待下一个调用信息，最后，客户端调用进程接收答复信息，获得进程结果，然后调用执行继续进行。

有多种 RPC模式和执行。最初由 Sun 公司提出。IETF ONC 宪章重新修订了 Sun 版本，使得 ONC RPC 协议成为 IETF 标准协议。现在使用最普遍的模式和执行是开放式软件基础的分布式计算环境（DCE）。

-------------百度百科

　　在简单了解了RPC的相关概念后，我们通过Java来开发一个简易版的RPC框架，所涉及的知识包括java的套接字（Socket）+java反射+JDK动态代理

常见的RPC实现由两种方式，一种为TCP（应用层为HTTP），一种为面向无连接的UDP。这里我们使用TCP来实现一个简易版的RPC框架

相关内容可以参考阿里大神（dubbo的开发者之一）的内容：http://javatar.iteye.com/blog/1123915

言归正传：所涉及的模块包括 1、RPC服务器 2、接口 3、实现类 4、服务器提供者模块 5、客户端消费者模块

1、RPC服务器+接口代理类

package com.jd.rpc;

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.ServerSocket;

import java.net.Socket;

/**

 * RpcFramework

 * 包含 RPC服务器+接口代理类

 *

 * @author william.liangf

 */

public class RpcFramework {

    /**

     * 暴露RPC服务

     *

     * @param service 服务实现

     * @param port 服务端口

     * @throws Exception

     */

    public static void export(final Object service, int port) throws Exception {

        if (service == null)

            throw new IllegalArgumentException("service instance == null");

        if (port <= 0 || port > 65535)//端口范围0~65535

            throw new IllegalArgumentException("Invalid port " + port);

        System.out.println("Export service " + service.getClass().getName() + " on port " + port);

        ServerSocket server = new ServerSocket(port);

        for(;;) {

            try {

                final Socket socket = server.accept();//启动服务端的Socket，等待客户端来连接

                new Thread(new Runnable() {//每次连接都启动一个全新的线程

                    @Override

                    public void run() {

                        try {

                            try {

                                ObjectInputStream input = new ObjectInputStream(socket.getInputStream());//获取socket传递过来的数据

                                try {

                                    String methodName = input.readUTF();//接口方法名

                                    Class<?>[] parameterTypes = (Class<?>[])input.readObject();//接口与参数类型

                                    Object[] arguments = (Object[])input.readObject();//具体参数值

                                    ObjectOutputStream output = new ObjectOutputStream(socket.getOutputStream());//获取回传output

                                    try {

                                        //通过java的反射，动态执行实现类service的方法

                                        Method method = service.getClass().getMethod(methodName, parameterTypes);

                                        Object result = method.invoke(service, arguments);

                                        //将调用结果回传给调用方

                                        output.writeObject(result);

                                    } catch (Throwable t) {

                                        output.writeObject(t);//关闭

                                    } finally {

                                        output.close();//关闭

                                    }

                                } finally {

                                    input.close();//关闭

                                }

                            } finally {

                                socket.close();//关闭

                            }

                        } catch (Exception e) {

                            e.printStackTrace();

                        }

                    }

                }).start();

            } catch (Exception e) {

                e.printStackTrace();

            }

        }

    }

    /**

     * 引用服务，即接口代理类

     *

     * @param <T> 接口泛型

     * @param interfaceClass 接口类型

     * @param host 服务器主机名

     * @param port 服务器端口

     * @return 远程服务

     * @throws Exception

     */

    @SuppressWarnings("unchecked")

    public static <T> T refer(final Class<T> interfaceClass, final String host, final int port) throws Exception {

        if (interfaceClass == null)

            throw new IllegalArgumentException("Interface class == null");

        if (! interfaceClass.isInterface())

            throw new IllegalArgumentException("The " + interfaceClass.getName() + " must be interface class!");

        if (host == null || host.length() == 0)

            throw new IllegalArgumentException("Host == null!");

        if (port <= 0 || port > 65535)

            throw new IllegalArgumentException("Invalid port " + port);

        System.out.println("Get remote service " + interfaceClass.getName() + " from server " + host + ":" + port);

        //创建JDK 动态代理类，作为接口的代理类

        //其中最主要的是在此接口代理类中 需要执行套接字的相关内容，以便发送相关的套接字内容

        //其中最主要的是在代理类中实现套接字请求

        return (T) Proxy.newProxyInstance(interfaceClass.getClassLoader(), new Class<?>[] {interfaceClass}, new InvocationHandler() {

            public Object invoke(Object proxy, Method method, Object[] arguments) throws Throwable {

                //套接字连接

                Socket socket = new Socket(host, port);

                try {

                    //对象输出流

                    ObjectOutputStream output = new ObjectOutputStream(socket.getOutputStream());

                    try {

                        output.writeUTF(method.getName());//传入参数 方法名称

                        output.writeObject(method.getParameterTypes());//传入参数 参数类型

                        output.writeObject(arguments);//传入参数 参数对象

                        ObjectInputStream input = new ObjectInputStream(socket.getInputStream());//接收RPC服务器回传回来的内容

                        try {

                            Object result = input.readObject();

                            if (result instanceof Throwable) {

                                throw (Throwable) result;

                            }

                            return result;//返回结果对象

                        } finally {

                            input.close();//关闭

                        }

                    } finally {

                        output.close();//关闭

                    }

                } finally {

                    socket.close();//关闭

                }

            }

        });

    }

}

2、定义接口服务

/**

 *  定义接口服务

 *

 * @author william.liangf

 */

public interface HelloService {

    String hello(String name);

}

3、接口服务实现类

package com.jd.rpc;

/**

 *接口服务实现类

 *

 * @author william.liangf

 */

public class HelloServiceImpl implements HelloService {

    public String hello(String name) {

        return "Hello " + name;

    }

}

3、RPC 服务提供者

package com.jd.rpc;

/**

 * RpcProvider

 * RPC 服务提供者 可以理解为生产者

 *

 * @author william.liangf

 */

public class RpcProvider {

    public static void main(String[] args) throws Exception {

        HelloService service = new HelloServiceImpl();

        RpcFramework.export(service, 1234);

    }

}

4、RPC服务调用者

package com.jd.rpc;

/**

 * RpcConsumer

 * RPC服务调用者 可以理解为消费者

 *

 * @author william.liangf

 */

public class RpcConsumer {

    public static void main(String[] args) throws Exception {

        HelloService service = RpcFramework.refer(HelloService.class, "127.0.0.1", 1234);

        for (int i = 0; i < Integer.MAX_VALUE; i ++) {

            String hello = service.hello("World" + i);

            System.out.println(hello);

            Thread.sleep(1000);

        }

    }

}

相关内容可以参考阿里大神（dubbo的开发者之一）的内容：http://javatar.iteye.com/blog/1123915

巴特西

RPC框架简易实现

最新文章

热门文章