手把手教你写一个RPC

1.1 rpc 是什么

定义：rpc（remote procedure call protocol）——远程过程调用协议 ，rpc协议假定某些传输协议的存在，如tcp或udp，为通信程序之间携带信息数据。在osi网络通信模型中，rpc跨越了传输层和应用层 ，rpc使得开发包括网络分布式多程序在内的应用程序更加容易。

我的理解：与其说把rpc 看作是一种协议，倒不如把 它看作是一种 客户机/服务器交互的模式，但是 rpc一定是基于 tcp 或者 其他 通信协议的

下面我们来看一下一个rpc调用的流程涉及哪些通信细节：

• 服务消费方（client）调用以本地调用方式调用服务；(1)
• client stub接收到调用后负责将方法、参数等组装成能够进行网络传输的消息体；(2)
• client stub找到服务地址，并将消息发送到服务端；(3)
• server stub收到消息后进行解码；(4)
• server stub根据解码结果调用本地的服务；(5)
• 本地服务执行并将结果返回给server stub；(6)
• server stub将返回结果打包成消息并发送至消费方；(7)
• client stub接收到消息，并进行解码；(8)
• 服务消费方得到最终结果。(9)

rpc的目标就是要2~8这些步骤都封装起来，让用户对这些细节透明。

1.2 手动实现

1.2.1 先做一个空接口实现序列化接口

public interface irpcservice extends serializable{

}

1.2.2 做一个需要被远程调用的接口 以及对应的接口实现类

public interface ihelloservice extends irpcservice{
    string sayhi(string name,string message); 
}

public class helloserviceimpl implements ihelloservice{

    private static final long serialversionuid = 146468468464364698l;

    @override
    public string sayhi(string name, string message) {
        return new stringbuilder().append("hi~!").append(",").append(message).tostring();
    }

}

1.2.3 需要写一个服务端，主要的作用 是进行服务注册(接口注册) 以及 接收客户端的调用参数 执行调用请求 返回结果

注：这个地方 我没有采用dom4j 解析配置文件的形式 进行接口注册 有时间的朋友可以多加一层

public interface server {
    
    //socket端口
    int port = 8080;
    
    //启动服务端
    void start() throws ioexception;
    
    //停止服务端
    void stop();
    
    /**
     * 服务注册
     * -- serviceinterface 对外暴露接口
     * -- 内部实现类
     */
    void regist(class<? extends irpcservice> serviceinterface,class<? extends irpcservice> impl);
    
}

public class servercenter implements server{
    
    /**线程池 接收客户端调用**/
    private static threadpoolexecutor executor = new threadpoolexecutor(5, 20, 200, timeunit.milliseconds,new arrayblockingqueue<runnable>(10));
    /**服务注册缓存**/
    public static final map<string,class<?>> serviceregistry = new hashmap<>();
    
    /**
     * 启动服务
     */
    @override
    public void start() throws ioexception {
        serversocket server = new serversocket();
        server.bind(new inetsocketaddress(port));
        try {
            while(true){
                executor.execute(new servicetask(server.accept()));
            }
        } finally {
            server.close();
        }
    }
    
    /**
     * 停止服务
     */
    @override
    public void stop() {
        executor.shutdown();
    }
    
    /**
     * 注册服务
     */
    @override
    public void regist(class<? extends irpcservice> serviceinterface, class<? extends irpcservice> impl) {
        serviceregistry.put(serviceinterface.getname(), impl);
    }
    
    private static class servicetask implements runnable{
        
        socket client = null;
        
        public servicetask(socket client) {
            this.client = client;
        }
        
        @override
        public void run() {
            objectinputstream input = null;
            objectoutputstream output = null;
            
            try {
                
                input = new objectinputstream(client.getinputstream());
                string servicename = input.readutf();
                string methodname = input.readutf();
                
                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);
                
                //将执行结果反序列化 通过socket返给客户端
                output = new objectoutputstream(client.getoutputstream());
                output.writeobject(result);
                
            } catch (exception e) {
                e.printstacktrace();
            } finally {
                
                if(input != null){
                    try {
                        input.close();
                    } catch (ioexception e) {
                        e.printstacktrace();
                    }
                }
                
                if(output != null){
                    try {
                        output.close();
                    } catch (ioexception e) {
                        e.printstacktrace();
                    }
                }
                
                if(client != null){
                    try {
                        client.close();
                    } catch (ioexception e) {
                        e.printstacktrace();
                    }
                }
                
            }
        }
        
    }
    
    public static void main(string[] args) throws exception {
        servercenter center = new servercenter();
        center.regist(ihelloservice.class,new helloserviceimpl().getclass());
        center.start();
    }
}

1.2.4 写一个客户端，用动态代理 获取被代理接口的 各种参数 传输给 服务端，接收返回结果，打印到控制台

public class client {
    
    
    @suppresswarnings("unchecked")
    public static <t extends irpcservice>t getremoteproxyobj(final class<? extends irpcservice> serviceinterface,final inetsocketaddress addr){
        return (t) proxy.newproxyinstance(serviceinterface.getclassloader(), new class<?>[]{serviceinterface}, new invocationhandler() {
            
            @override
            public object invoke(object proxy, method method, object[] args) throws throwable {
                socket socket = null;
                objectoutputstream output = null;
                objectinputstream input = null;
                try {
                    //1.创建socket客户端，根据指定地址连接远程服务提供者
                    socket = new socket();
                    socket.connect(addr);
                    
                    //2.将远程服务调用所需的接口类、方法名、参数列表等编码后发送给服务提供者
                    output = new objectoutputstream(socket.getoutputstream());
                    output.writeutf(serviceinterface.getname());
                    output.writeutf(method.getname());
                    output.writeobject(method.getparametertypes());
                    output.writeobject(args);
                    
                    //3.同步阻塞等待服务器返回应答 获取应答后返回
                    input = new objectinputstream(socket.getinputstream());
                    return input.readobject();
                } finally{
                    if(socket != null){
                        socket.close();
                    }
                    
                    if(output != null){
                        output.close();
                    }
                    
                    if(input != null){
                        input.close();
                    }
                }
            }
        });
    }
    
}

1.2.5 测试

注：测试之前 需要开启服务端

public class rpctest {
    public static void main(string[] args) throws ioexception {
        ihelloservice service  = client.getremoteproxyobj(ihelloservice.class, new inetsocketaddress(8080));
        system.out.println(service.sayhi("张三", "新年快乐!"));
    }
}

就这样我们实现了一个简陋的rpc

本文意在通过实现简单的rpc，去真正意义上对rpc框架的实现原理有初步的了解，而不是人云亦云。

此rpc实现有诸多缺点，但是 我们只要明白rpc的基座 其他的rpc框架只是完善基座以及扩展而已 。