Hadoop RPC简单例子

Posted 2020-09-10 lnlvinso

　　jdk中已经提供了一个RPC框架—RMI，但是该PRC框架过于重量级并且可控之处比较少，所以Hadoop RPC实现了自定义的PRC框架。

　

　　同其他RPC框架一样，Hadoop RPC分为四个部分：

　　（1）序列化层：Clent与Server端通信传递的信息采用了Hadoop里提供的序列化类或自定义的Writable类型；

　　（2）函数调用层：Hadoop RPC通过动态代理以及java反射实现函数调用；

　　（3）网络传输层：Hadoop RPC采用了基于TCP/IP的socket机制；

　　（4）服务器端框架层：RPC Server利用java NIO以及采用了事件驱动的I/O模型，提高RPC Server的并发处理能力；

 

       下面是基于Hadoop2.6版本的简单例子

　　 一、定义接口  

public interface IProxyProtocol extends VersionedProtocol {
    static final long versionID= 23234L; //版本号，默认情况下，不同版本号的RPC Client和Server之间不能相互通信
    int Add(int number1,int number2);
}

　　二、实现接口

public class MyProxy implements IProxyProtocol {
    public int Add(int number1,int number2) {
        System.out.println("我被调用了!");
        int result = number1+number2;
        return result;
    }

    public long getProtocolVersion(String protocol, long clientVersion)throws IOException {

        System.out.println("MyProxy.ProtocolVersion=" + IProxyProtocol.versionID); 

        // 注意：这里返回的版本号与客户端提供的版本号需保持一致 return IProxyProtocol.VERSION; 
    } 
  
    public ProtocolSignature getProtocolSignature(String arg0,long arg1,int arg2) throws IOException{ 
 
         return new ProtocolSignature(); } 
   }

　　三、服务器端代码

 

public class MyServer {
    public static int PORT = 5432;
    public static String IPAddress = "127.0.0.1";

    public MyServer (){
        MyProxy proxy = new MyProxy();
        RPC.Builder ins = new RPC.Builder(new Configuration);
        ins.setInstance(proxy);
        ins.setBindAddress(IPAddress);
        ins.setPort(PORT);
        ins.setProtocol(MyProxy.class);
        Server server;
        try{
             server = ins.build();
             server.start();
         }catch(Exception e){
             e.printStackTrace();
         }
    }

    public static void main(String[] args) throws Exception {
        MyProxy proxy = new MyProxy();
    }
}

　　四、客户端代码

public class MyClient {

    public static void main(String[] args) {
        InetSocketAddress inetSocketAddress = new InetSocketAddress(
                MyServer.IPAddress, MyServer.PORT);

        try {
            // 注意：这里传入的版本号需要与代理保持一致
            IProxyProtocol proxy = (IProxyProtocol) RPC.waitForProxy(
                    IProxyProtocol.class, IProxyProtocol.versionID, inetSocketAddress,
                    new Configuration());
            int result = proxy.Add(10, 25);
            System.out.println("10+25=" + result);

            RPC.stopProxy(proxy);
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

}

 

　　先启动服务端代码，再启动客户端代码。