码迷,mamicode.com
首页 > 其他好文 > 详细

Hadoop RPC远程过程调用源码解析及实例

时间:2015-06-08 21:37:20      阅读:169      评论:0      收藏:0      [点我收藏+]

标签:hadoop   rpc   源码解析   实例   

什么是RPC?

1、RPC(Remote Procedure Call)远程过程调用,它允许一台计算机程序远程调用另外一台计算机的子程序,而不用去关心底层的网络通信细节,对我们来说是透明的。经常用于分布式网络通信中。
2、Hadoop的进程间交互都是通过RPC来进行的,比如Namenode与Datanode之间,Jobtracker与Tasktracker之间等。

RPC协议假定某些传输协议的存在,如TCP或UDP,为通信程序之间携带信息数据。在OSI网络通信模型中, RPC跨越了传输层和应用层。 RPC使得开发包括网络分布式多程序在内的应用程序更加容易。

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

RPC特点

1、透明性:远程调用其他机器上的程序,对用户来说就像是调用本地方法一样。
2、高性能:RPC server能够并发处理多个来自Client的请求(请求队列)。
3、可控性:jdk中已经提供了一个RPC框架–RMI,但是该RPC框架过于重量级并且可控之处比较少,所以Hadoop RPC实现了自定义的RPC框架。

Hadoop RPC通信

1、序列化层:Client与Server端 通信传递的信息采用了Hadoop里提供的序列化类或自定义Writable类型。

2、函数调用层:Hadoop RPC通过动态代理以及Java反射机制实现函数调用。

3、网络传输层:Hadoop RPC采用了基于TCP/IP的socket机制。

4、服务器端框架层:RPC Server利用Java NIO以及采用了事件驱动的I/O模型,提高RPC Server的并发处理能力

Hadoop的整个体系结构就是构建在RPC之上(org.apache.hadoop.ipc)。

Hadoop RPC设计技术

1、动态代理
2、反射
3、序列化
4、非阻塞的异步IO(NIO)

动态代理

1、动态代理可以提供对另一个对象的访问,同时隐藏实际对象的具体事实,代理对象对客户隐藏了实际对象。
2、动态代理可以对请求进行其他的一些处理,在不允许直接访问某些类,或需要对访问做一些特殊处理等,这时候可以考虑使用代理。
3)目前Java开发包中提供了对动态代理的支持,但现在只支持对接口的实现。
相关的类与接口:

java.lang.reflect.Proxy   --类
java.lang.reflect.InvocationHandler  --接口

动态代理创建对象过程:

InvocationHandler  handler = new InvocationHandlerImpl(...)
Proxy.newInstance(...)

具体实现可参考如下:
http://blog.csdn.net/scgaliguodong123_/article/details/44025531

技术分享

根据上图查看hadoop2.6.0源码

Client
技术分享

Server
技术分享

RPC
技术分享

几个重要的协议

ClientProtocol是客户端(FileSystem)与NameNode通信的接口。
DatanodeProtocol是DataNode与NameNode通信的接口
NamenodeProtocol是SecondaryNameNode与NameNode
通信的接口。
DFSClient是直接调用NameNode接口的对象。用户代码是通过DistributedFileSystem调用DFSClient对象,才能与NameNode打交道。

技术分享

技术分享

模拟Hadoop RPC通信

package MyRPC;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.ipc.VersionedProtocol;

public interface MyRPCProtocal extends VersionedProtocol{
    public static long versionID = 23234l;//很重要很重要,搞了一下午才解决掉。
    public Text test(Text t);
}
package MyRPC;

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.ipc.ProtocolSignature;
import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.ipc.RPC.Server;

public class RPCServer implements MyRPCProtocal{    
    Server server = null;
    public RPCServer() throws IOException, InterruptedException{
        //server = RPC.getServer(this,"localhost",8888,new Configuration());
        //相对于以前的版本有略微的改动
        RPC.Builder ins = new RPC.Builder(new Configuration());
        ins.setInstance(this);
        ins.setBindAddress("localhost");
        ins.setPort(9999);
        ins.setProtocol(MyRPCProtocal.class);
        //RPC.setProtocolEngine(new Configuration(), MyRPCProtocal.class, RpcEngine.class);
        server = ins.build();//获得一个server实例
        server.start();
        server.join();  
    }

    public static void main(String[] args) throws IOException, InterruptedException {
        new RPCServer();
    }

    @Override
    public long getProtocolVersion(String protocol, long clientVersion)
            throws IOException {
        return MyRPCProtocal.versionID;
    }

    @Override
    public ProtocolSignature getProtocolSignature(String protocol,
            long clientVersion, int clientMethodsHash) throws IOException {     
        return new ProtocolSignature();
    }

    @Override
    public Text test(Text t) {
        if(t.toString().equals("RPC")){
            return new Text("ok");
        }
        return new Text("false");
    }
}
package MyRPC;

import java.net.InetSocketAddress;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.ipc.RPC;

public class RPCClient {

    private MyRPCProtocal protocal;

    public RPCClient() throws Exception{
        InetSocketAddress address = new InetSocketAddress("localhost",9999);

        protocal = (MyRPCProtocal)RPC.waitForProxy
                (MyRPCProtocal.class,MyRPCProtocal.versionID, address, new Configuration());
        //RPC.setProtocolEngine(new Configuration(), MyRPCProtocal.class, RpcEngine.class);
    }

    public void call(String s){
        final Text string = protocal.test(new Text(s));
        System.out.println(string.toString());
    }

    public static void main(String[] args) throws Exception {
        RPCClient client = new RPCClient();
        client.call("RPC");
    }
}

Hadoop RPC远程过程调用源码解析及实例

标签:hadoop   rpc   源码解析   实例   

原文地址:http://blog.csdn.net/scgaliguodong123_/article/details/46405093

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!