`
flyfoxs
  • 浏览: 297697 次
  • 性别: Icon_minigender_1
  • 来自: 合肥
社区版块
存档分类
最新评论

【大数据笔记】Hadoop通过动态代理实现RPC

阅读更多
Hadoop所有的跨节点的通信都是通过RPC来通信的, RPC通信是需要创建Stub,一个好的RPC需要通过良好的设计确保了对上层调用的透明性. 我们下面就通过Hadoop里面最常用的心跳(JobTrack和TaskTracker)来研究一下Hadoop的RPC机制.
心跳函数的调用在方法:TaskTracker.transmitHeartBeat();在此方法中会调用下面的代码段.这个方法实际就不一个本地调用,jobClient实际上就是一个动态代理生成的对象,这个对象已经包括了RPC的Stub.
    //
    // Xmit the heartbeat
    //
    HeartbeatResponse heartbeatResponse = jobClient.heartbeat(status, 
                                                              justStarted,
                                                              justInited,
                                                              askForNewTask, 
                                                              heartbeatResponseId);
 
也就是 TaskTracker 通过 jobClient(InterTrackerProtocol) 来调用 JobTracker, 而JobClient实现了接口 jobClient(InterTrackerProtocol).
这是一个典型的Java动态代理的应用场景. 同时RPC调用需要跨越网络,现在我们就来看看Hadoop是如何透明的创建一个jobClient实例.
 
//这个是对jobClient的定义
InterTrackerProtocol jobClient;


//下面可以看出jobClient和JobTracker 实现了相同的接口
interface InterTrackerProtocol extends VersionedProtocol;


public class JobTracker implements MRConstants, InterTrackerProtocol,
    JobSubmissionProtocol, TaskTrackerManager, RefreshUserMappingsProtocol,
    RefreshAuthorizationPolicyProtocol, AdminOperationsProtocol,
    JobTrackerMXBean {
  
//下面是jobClient初始化的代码
this.jobClient = (InterTrackerProtocol)
    UserGroupInformation.getLoginUser().doAs(
        new PrivilegedExceptionAction<Object>() {
      public Object run() throws IOException {
        return RPC.waitForProxy(InterTrackerProtocol.class,
            InterTrackerProtocol.versionID,
            jobTrackAddr, fConf);
      }
    });
 
下图就是来自于对TaskTracker.initialize()的跟踪, 从该调用栈就可以看出TaskTracker是如何通过动态代理创建出一个RPC 客户端, 却不需要调用者提供任何信息.
2点注意:
1)RPC客户端如何得到端口号,服务器地址 
下面的代码已经添加了注释,可以看到,RPC客户端如何得到服务器地址,端口号,并生成stub,形成动态代理.
//下面是Java动态代理调用处, RPC.getProxy()
//参数addr已经封装了端口号,和地址, 这个地址就是TaskTracker里面的成员变量jobTrackAddr一级一级传递下来,传递路径就是上图所示.
    VersionedProtocol proxy =
        (VersionedProtocol) Proxy.newProxyInstance(
            protocol.getClassLoader(), new Class[] { protocol },
            new Invoker(protocol, addr, ticket, conf, factory, rpcTimeout));


//最后一个参数Invoker是RPC的一个内部类,它实现了接口InvocationHandler,下面是他的构造函数
private static class Invoker implements InvocationHandler {
    private Client.ConnectionId remoteId;
    private Client client;
    private boolean isClosed = false;
//在构造函数中,客户端和服务器建立连接
    public Invoker(Class<? extends VersionedProtocol> protocol,
        InetSocketAddress address, UserGroupInformation ticket,
        Configuration conf, SocketFactory factory,
        int rpcTimeout) throws IOException {
      this.remoteId = Client.ConnectionId.getConnectionId(address, protocol,
          ticket, rpcTimeout, conf);
      this.client = CLIENTS.getClient(conf, factory);
    }
//调用时,直接把准备好的参数通过连接发给服务器并同步获取返回值
    public Object invoke(Object proxy, Method method, Object[] args)
      throws Throwable {
      final boolean logDebug = LOG.isDebugEnabled();
      long startTime = 0;
      if (logDebug) {
        startTime = System.currentTimeMillis();
      }
      ObjectWritable value = (ObjectWritable)
        client.call(new Invocation(method, args), remoteId);
      if (logDebug) {
        long callTime = System.currentTimeMillis() - startTime;
        LOG.debug("Call: " + method.getName() + " " + callTime);
      }
      return value.get();
    }
   
    /* close the IPC client that's responsible for this invoker's RPCs */
    synchronized private void close() {
      if (!isClosed) {
        isClosed = true;
        CLIENTS.stopClient(client);
      }
    }
  }
  
2)服务器采用的是NIO,客户端采用的是BIO
客户端的调用就是BIO,不是NIO.调用方式在上文已经通过jobClient.heartbeat()举例做了相应的分析.
Listener将以独立的线程启动,并以异步(NIO)方式监听指定端口.下面的代码已经添加了注释,可以看出Listener如何监听连接,并开启对应的Reader来异步处理对应的数据.
Listener服务开启于JobTracker,因为要面对大量的TaskTracker连接,所以使用Server使用NIO来处理,这也是NIO典型的应用场景.
//Listener是  org.apache.hadoop.ipc.Server的内部类(实际上是Server的父类的内部类)

//下面是Listener唯一的构造函数
    public Listener() throws IOException {
      address = new InetSocketAddress(bindAddress, port);
      // Create a new server socket and set to non blocking mode
      acceptChannel = ServerSocketChannel.open();
      acceptChannel.configureBlocking(false);
      // Bind the server socket to the local host and port
      bind(acceptChannel.socket(), address, backlogLength);
      port = acceptChannel.socket().getLocalPort(); //Could be an ephemeral port
      // create a selector;
      selector= Selector.open();
      readers = new Reader[readThreads];
      readPool = Executors.newFixedThreadPool(readThreads);
      for (int i = 0; i < readThreads; i++) {
        Selector readSelector = Selector.open();
        Reader reader = new Reader(readSelector);
        readers[i] = reader;
        //并不会真正的开始处理数据,Reader在没有被Linster初始化之前,run()会一直处于自旋等待
        readPool.execute(reader);
      }
      // Register accepts on the server socket with the selector.
      //绑定selector和连接初始化事件,连接初始化时,在doAccept里面先对Reader初始化,然后Reader开始处理数据.
      acceptChannel.register(selector, SelectionKey.OP_ACCEPT);
      this.setName("IPC Server listener on " + port);
      this.setDaemon(true);
    }
 
 
 参考:http://blog.csdn.net/jiangwei0910410003/article/details/21155911
  • 大小: 45.1 KB
0
0
分享到:
评论

相关推荐

    基于springboot教育资源共享平台源码数据库文档.zip

    基于springboot教育资源共享平台源码数据库文档.zip

    视频笔记linux开发篇

    linux开发篇,配套视频:https://www.bilibili.com/list/474327672?sid=4493702&spm_id_from=333.999.0.0&desc=1

    readera-24-09-08plus2020.apk

    ReadEra 这个阅读应用能够打开下列任何格式的文档: EPUB, PDF, DOC, RTF, TXT, DJVU, FB2, MOBI, 和 CHM. 基本上来说,你可以用它阅读你的设备内存中的任何书籍或者文本文档。 这个应用与划分成章节的文档兼。,有一个书签功能,可以在你阅读的时候,自动保存你的进度。另外,它让你更改页面模式,从几种不同的主题中进行挑选(夜间,白天,棕黑色调,还有控制台)。

    STM32单片机控制舵机旋转

    软件环境:KEIL4 硬件环境:STM32单片机+舵机 控制原理:通过控制输出信号的占空比调节舵机旋转的角度

    基于springboot仓库管理系统源码数据库文档.zip

    基于springboot仓库管理系统源码数据库文档.zip

    酒店管理系统源码C++实现的毕业设计项目源码.zip

    酒店管理系统源码C++实现的毕业设计项目源码.zip,个人大四的毕业设计、经导师指导并认可通过的高分设计项目,评审分98.5分。主要针对计算机相关专业的正在做毕设的学生和需要项目实战练习的学习者,也可作为课程设计、期末大作业。 酒店管理系统源码C++实现的毕业设计项目源码.zip,酒店管理系统源码C++实现的毕业设计项目源码.zip个人大四的毕业设计、经导师指导并认可通过的高分设计项目,评审分98.5分。主要针对计算机相关专业的正在做毕设的学生和需要项目实战练习的学习者,也可作为课程设计、期末大作业。酒店管理系统源码C++实现的毕业设计项目源码.zip酒店管理系统源码C++实现的毕业设计项目源码.zip酒店管理系统源码C++实现的毕业设计项目源码.zip,个人大四的毕业设计、经导师指导并认可通过的高分设计项目,评审分98.5分。主要针对计算机相关专业的正在做毕设的学生和需要项目实战练习的学习者,也可作为课程设计、期末大作业。酒店管理系统源码C++实现的毕业设计项目源码.zip,个人大四的毕业设计、经导师指导并认可通过的高分设计项目,评审分98.5分。主要针对计算机相关专业的正在做毕

    58商铺全新UI试客试用平台网站源码

    58商铺全新UI试客试用平台网站源码

    基于SpringBoot+Vue的轻量级定时任务管理系统.zip

    springboot vue3前后端分离 基于SpringBoot+Vue的轻量级定时任务管理系统.zip

    毕业设计&课设_微博情感分析,用 flask 构建 restful api,含相关算法及数据文件.zip

    该资源内项目源码是个人的课程设计、毕业设计,代码都测试ok,都是运行成功后才上传资源,答辩评审平均分达到96分,放心下载使用! ## 项目备注 1、该资源内项目代码都经过严格测试运行成功才上传的,请放心下载使用! 2、本项目适合计算机相关专业(如计科、人工智能、通信工程、自动化、电子信息等)的在校学生、老师或者企业员工下载学习,也适合小白学习进阶,当然也可作为毕设项目、课程设计、作业、项目初期立项演示等。 3、如果基础还行,也可在此代码基础上进行修改,以实现其他功能,也可用于毕设、课设、作业等。 下载后请首先打开README.md文件(如有),仅供学习参考, 切勿用于商业用途。

    4D毫米波雷达点云数据处理方法研究.caj

    4D毫米波雷达点云数据处理方法研究.caj

    S M 2 2 5 8 X T量产工具

    S M 2 2 5 8 X T 量产工具供大家下载使用

    基于springboot的文物管理系统源码数据库文档.zip

    基于springboot的文物管理系统源码数据库文档.zip

    基于springboot的电影院售票管理系统源码数据库文档.zip

    基于springboot的电影院售票管理系统源码数据库文档.zip

    Javaweb仓库管理系统项目源码.zip

    基于Java web 实现的仓库管理系统源码,适用于初学者了解Java web的开发过程以及仓库管理系统的实现。

    美容美发项目,使用django框架,前后端一体化项目

    美容美发项目,使用django框架,前后端一体化项目

    2023年中国在线票务行业市场规模约为24.99亿元,挖掘市场新机遇

    在线票务:2023年中国在线票务行业市场规模约为24.99亿元,挖掘市场蓝海新机遇 在数字浪潮的席卷下,传统的票务销售模式正经历着前所未有的变革。纸质门票逐渐淡出人们的视野,取而代之的是便捷、高效的数字和移动票务。这一转变不仅为消费者带来了前所未有的购票体验,更为在线票务平台开辟了广阔的发展空间和市场机遇。随着国民经济的持续增长和文体娱乐行业的蓬勃发展,中国在线票务行业正站在时代的风口浪尖,等待着每一位有志之士的加入。那么,这片蓝海市场究竟蕴藏着怎样的潜力?又该如何把握机遇,实现突破?让我们一同探索。 市场概况: 近年来,中国在线票务行业市场规模持续扩大,展现出强劲的增长势头。据QYResearch数据显示,2023年中国在线票务行业市场规模约为24.99亿元,尽管受到宏观经济的影响,市场规模增速放缓,但整体趋势依然向好。这一增长主要得益于国民人均收入的不断提高、电影及演出行业的快速发展以及政府政策的支持。例如,2023年财政部、国家电影局发布的《关于阶段性免征国家电影事业发展专项资金政策的公告》,为电影行业注入了强劲动力,进而推动了在线票务市场规模的扩大。 技术创新与趋势: 技术进步

    基于SpringBoot的养老院管理系统源码数据库文档.zip

    基于SpringBoot的养老院管理系统源码数据库文档.zip

    毕业设计&课设_含构建设置及相关操作,基于特定技术,具体功能未详细说明.zip

    该资源内项目源码是个人的课程设计、毕业设计,代码都测试ok,都是运行成功后才上传资源,答辩评审平均分达到96分,放心下载使用! ## 项目备注 1、该资源内项目代码都经过严格测试运行成功才上传的,请放心下载使用! 2、本项目适合计算机相关专业(如计科、人工智能、通信工程、自动化、电子信息等)的在校学生、老师或者企业员工下载学习,也适合小白学习进阶,当然也可作为毕设项目、课程设计、作业、项目初期立项演示等。 3、如果基础还行,也可在此代码基础上进行修改,以实现其他功能,也可用于毕设、课设、作业等。 下载后请首先打开README.md文件(如有),仅供学习参考, 切勿用于商业用途。

    Go语言入门指南:基础语法、并发编程详解

    内容概要:本文档是一份详细的Go语言教程,从基础概念介绍到高级主题均有覆盖。主要内容包括Go语言的基础语法、数据类型、控制结构、函数、结构体、接口和并发编程等方面。通过具体示例介绍了如何使用Go语言进行开发。 适合人群:初学者和有一定经验的程序员都可以从这篇教程中受益,特别是那些想要快速掌握Go语言并应用于实际项目的开发者。 使用场景及目标:适用于初学者系统学习Go语言的基础知识和常用功能;也可以作为已有开发经验者的参考资料,帮助他们解决具体的编程问题,提高开发效率。 其他说明:本教程不仅包含了Go语言的基本知识点,还重点讲解了其独特的并发编程模型。读者在学习过程中应该注重理论与实践相结合,通过实际编写代码来加深理解和记忆。

    基于springboot计算机基础网上考试系统源码数据库文档.zip

    基于springboot计算机基础网上考试系统源码数据库文档.zip

Global site tag (gtag.js) - Google Analytics