Hadoop2源码分析-RPC机制初识-阿里云开发者社区

开发者社区> 开发与运维> 正文

Hadoop2源码分析-RPC机制初识

简介:

1.概述

  上一篇博客,讲述Hadoop V2的序列化机制,这为我们学习Hadoop V2的RPC机制奠定了基础。RPC的内容涵盖的信息有点多,包含Hadoop的序列化机制,RPC,代理,NIO等。若对Hadoop序列化不了解的同学,可以参考《Hadoop2源码分析-序列化篇》。今天这篇博客为大家介绍的内容目录如下:

  • RPC概述
  • 第三方RPC
  • Hadoop V2的RPC简述

  那么,下面开始今天的学习之路。

2.RPC概述

  首先,我们要弄明白,什么是RPC?RPC能用来做什么?

2.1什么是RPC

  RPC的全程是Remote Procedure Call,中文释为远程过程调用。也就是说,调用的过程代码(业务服务代码)并不在调用者本地运行,而是要实现调用着和被调用着之间的连接通信,有同学可能已经发现,这个和C/S模式很像。没错,RPC的基础通信模式是基于C/S进程间相互通信的模式来实现的,它对Client端提供远程接口服务,其RPC原理图如下所示:

2.2RPC的功能

  我们都知道,在过去的编程概念中,过程是由开发人员在本地编译完成的,并且只能局限在本地运行的某一段代码,即主程序和过程程序是一种本地调用关系。因此,这种结构在如今网络飞速发展的情况下已无法适应实际的业务需求。而且,传统过程调用模式无法充分利用网络上其他主机的资源,如CPU,内存等,也无法提高代码在Bean之间的共享,使得资源浪费较大。

  而RPC的出现,正好有效的解决了传统过程中存在的这些不足。通过RPC,我们可以充分利用非共享内存的机器,可以简便的将应用分布在多台机器上,类似集群分布。这样方便的实现过程代码共享,提高系统资源的利用率。减少单个集群的压力,实现负载均衡。

3.第三方RPC

  在学习Hadoop V2的RPC机制之前,我们先来熟悉第三方的RPC机制是如何工作的,下面我以Thrift框架为例子。

  Thrift是一个软件框架,用来进行可扩展且跨语言的服务开发协议。它拥有强大的代码生成引擎,支持C++,Java,Python,PHP,Ruby等编程语言。Thrift允许定义一个简单的定义文件(以.thirft结尾),文件中包含数据类型和服务接口。用以作为输入文件,编译器生成代码用来方便的生成RPC客户端和服务端通信的编程语言。具体Thrift安装过程请参考《Mac OS X 下搭建thrift环境》。

3.1Thrift原理图

  下面给出Thrift的原理图,如下所示:

  下面为大家解释一下上面的原理图,首先,我们编译完thrift定义文件后(这里我使用的是Java语言),会生成对应的Java类文件,该类的Iface接口定义了我们所规范的接口函数。在服务端,实现Iface接口,编写对应函数下的业务逻辑,启动服务。客户端同样需要生成的Java类文件,以供Client端调用相应的接口函数,监听服务端的IP和PORT来获取连接对象。

3.2代码示例

  • Server端代码:
复制代码
package cn.rpc.main;

import org.apache.thrift.TProcessorFactory;
import org.apache.thrift.protocol.TCompactProtocol;
import org.apache.thrift.server.THsHaServer;
import org.apache.thrift.server.TServer;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TNonblockingServerSocket;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.rpc.service.StatQueryService;
import cn.rpc.service.impl.StatQueryServiceImpl;

/**
 * @Date Mar 23, 2015
 *
 * @Author dengjie
 */
public class StatsServer {

    private static Logger logger = LoggerFactory.getLogger(StatsServer.class);

    private final int PORT = 9090;

    @SuppressWarnings({ "rawtypes", "unchecked" })
    private void start() {
        try {
            TNonblockingServerSocket socket = new TNonblockingServerSocket(PORT);
            final StatQueryService.Processor processor = new StatQueryService.Processor(new StatQueryServiceImpl());
            THsHaServer.Args arg = new THsHaServer.Args(socket);
            /*
             * Binary coded format efficient, intensive data transmission, The
             * use of non blocking mode of transmission, according to the size
             * of the block, similar to the Java of NIO
             */
            arg.protocolFactory(new TCompactProtocol.Factory());
            arg.transportFactory(new TFramedTransport.Factory());
            arg.processorFactory(new TProcessorFactory(processor));
            TServer server = new THsHaServer(arg);
            server.serve();
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }

    public static void main(String[] args) {
        try {
            logger.info("start thrift server...");
            StatsServer stats = new StatsServer();
            stats.start();
        } catch (Exception ex) {
            ex.printStackTrace();
            logger.error(String.format("run thrift server has error,msg is %s", ex.getMessage()));
        }
    }

}
复制代码
  • Client端代码:
复制代码
package cn.rpc.test;

import java.util.Map;

import org.apache.thrift.protocol.TCompactProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;

import cn.rpc.service.StatQueryService;

/**
 * @Date Mar 23, 2015
 *
 * @Author dengjie
 * 
 * @Note Test thrift client
 */
public class StatsClient {

    public static final String ADDRESS = "127.0.0.1";
    public static final int PORT = 9090;
    public static final int TIMEOUT = 30000;

    public static void main(String[] args) {
        if (args.length < 4) {
            System.out.println("args length must >= 4,current length is " + args.length);
            System.out.println("<info>****************</info>");
            System.out.println("ADDRESS,beginDate,endDate,kpiCode,...");
            System.out.println("<info>****************</info>");
            return;
        }
        TTransport transport = new TFramedTransport(new TSocket(args[0], PORT, TIMEOUT));
        TProtocol protocol = new TCompactProtocol(transport);
        StatQueryService.Client client = new StatQueryService.Client(protocol);
        String beginDate = args[1]; // "20150308"
        String endDate = args[2]; // "20150312"
        String kpiCode = args[3]; // "login_times"
        String userName = "";
        int areaId = 0;
        String type = "";
        String fashion = "";

        try {
            transport.open();
            Map<String, String> map = client.queryConditionDayKPI(beginDate, endDate, kpiCode, userName, areaId, type,
                    fashion);
            System.out.println(map.toString());
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            transport.close();
        }
    }

}
复制代码
  • StatQueryService类:

  这个类的代码量太大,暂不贴出。需要的同学请到以下地址下载。

  下载地址:git@gitlab.com:dengjie/Resource.git

  • StatQueryServiceImpl类:

  下面实现其中一个函数的内容,代码如下所示:

复制代码
package cn.rpc.service.impl;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.thrift.TException;

import cn.rpc.conf.ConfigureAPI;
import cn.rpc.dao.KpiDao;
import cn.rpc.domain.ReportParam;
import cn.rpc.domain.ReportResult;
import cn.rpc.service.StatQueryService;
import cn.rpc.util.MapperFactory;

/**
 * @Date Mar 23, 2015
 *
 * @Author dengjie
 */
public class StatQueryServiceImpl implements StatQueryService.Iface {

    public Map<String, String> queryDayKPI(String beginDate, String endDate, String kpiCode) throws TException {
        return null;
    }

    public Map<String, String> queryConditionDayKPI(String beginDate, String endDate, String kpiCode, String userName,
            int areaId, String type, String fashion) throws TException {
        Map<String, String> res = new HashMap<String, String>();
        ReportParam param = new ReportParam();
        param.setBeginDate(beginDate + "");
        param.setEndDate(endDate + "");
        param.setKpiCode(kpiCode);
        param.setUserName(userName == "" ? null : userName);
        param.setDistrictId(areaId < 0 ? 0 : areaId);
        param.setProductStyle(fashion == "" ? null : fashion);
        param.setCustomerProperty(type == "" ? null : type);
        List<ReportResult> chart = ((KpiDao) MapperFactory.createMapper(KpiDao.class)).getChartAmount(param);
        Map<String, Integer> title = ((KpiDao) MapperFactory.createMapper(KpiDao.class)).getTitleAmount(param);
        List<Map<String, Integer>> tableAmount = ((KpiDao) MapperFactory.createMapper(KpiDao.class))
                .getTableAmount(param);
        String avgTime = kpiCode.split("_")[0];
        param.setKpiCode(avgTime + "_avg_time");
        List<Map<String, Integer>> tableAvgTime = ((KpiDao) MapperFactory.createMapper(KpiDao.class))
                .getTableAmount(param);
        res.put(ConfigureAPI.RESMAPKEY.CHART, chart.toString());
        res.put(ConfigureAPI.RESMAPKEY.TITLE, title.toString());
        res.put(ConfigureAPI.RESMAPKEY.TABLEAMOUNT, tableAmount.toString());
        res.put(ConfigureAPI.RESMAPKEY.TABLEAVG, tableAvgTime.toString());
        return res;
    }

    public Map<String, String> queryDetail(String beginDate, String endDate, String userName) throws TException {
        // TODO Auto-generated method stub
        return null;
    }

}
复制代码

4.Hadoop V2的RPC简述

  Hadoop V2中的RPC采用的是自己独立开发的协议,其核心内容包含服务端,客户端,交互协议。源码内容都在hadoop-common-project项目的org.apache.hadoop.ipc包下面。

  • VersionedProtocol类:
复制代码
package org.apache.hadoop.ipc;

import java.io.IOException;

/**
 * Superclass of all protocols that use Hadoop RPC.
 * Subclasses of this interface are also supposed to have
 * a static final long versionID field.
 */
public interface VersionedProtocol {
  
  /**
   * Return protocol version corresponding to protocol interface.
   * @param protocol The classname of the protocol interface
   * @param clientVersion The version of the protocol that the client speaks
   * @return the version that the server will speak
   * @throws IOException if any IO error occurs
   */
  public long getProtocolVersion(String protocol,
                                 long clientVersion) throws IOException;

  /**
   * Return protocol version corresponding to protocol interface.
   * @param protocol The classname of the protocol interface
   * @param clientVersion The version of the protocol that the client speaks
   * @param clientMethodsHash the hashcode of client protocol methods
   * @return the server protocol signature containing its version and
   *         a list of its supported methods
   * @see ProtocolSignature#getProtocolSignature(VersionedProtocol, String, 
   *                long, int) for a default implementation
   */
  public ProtocolSignature getProtocolSignature(String protocol, 
                                 long clientVersion,
                                 int clientMethodsHash) throws IOException;
}
复制代码

  该类中的两个方法一个是作为版本,另一个作为签名用。

  • RPC下的Server类:
复制代码
 /** An RPC Server. */
  public abstract static class Server extends org.apache.hadoop.ipc.Server {
   boolean verbose;
   static String classNameBase(String className) {
      String[] names = className.split("\\.", -1);
      if (names == null || names.length == 0) {
        return className;
      }
      return names[names.length-1];
    }
复制代码

  对外提供服务,处理Client端的请求,并返回处理结果。

  至于Client端,监听Server端的IP和PORT,封装请求数据,并接受Response。

5.总结

  这篇博客赘述了RPC的相关内容,让大家先熟悉一下RPC的相关机制和流程,并简述了Hadoop V2的RPC机制,关于Hadoop V2的RPC详细内容会在下一篇博客中给大家分享。这里只是让大家先对Hadoop V2的RPC机制有个初步的认识。

6.结束语

  这篇博客就和大家分享到这里,如果大家在研究学习的过程当中有什么问题,可以加群进行讨论或发送邮件给我,我会尽我所能为您解答,与君共勉!

联系方式: 
邮箱:smartloli.org@gmail.com 
Twitter:https://twitter.com/smartloli 
QQ群(Hadoop - 交流社区1):424769183 
温馨提示:请大家加群的时候写上加群理由(姓名+公司/学校),方便管理员审核,谢谢! 

热爱生活,享受编程,与君共勉!



本文转自哥不是小萝莉博客园博客,原文链接:http://www.cnblogs.com/smartloli/,如需转载请自行联系原作者

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

分享:
开发与运维
使用钉钉扫一扫加入圈子
+ 订阅

集结各类场景实战经验,助你开发运维畅行无忧

其他文章