AI 助理
备案控制台
开发者社区 大数据 文章 正文

Hadoop2源码分析－RPC探索实战

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

1.概述

　　在《Hadoop2源码分析－RPC机制初识》博客中，我们对RPC机制有了初步的认识和了解，下面我们对Hadoop V2的RPC机制做进一步探索，在研究Hadoop V2的RPC机制，我们需要掌握相关的Java基础知识，如：Java NIO、动态代理与反射等。本篇博客介绍的内容目录如下所示：

  • Java NIO简述
  • Java NIO实例演示
  • 动态代理与反射简述
  • 动态代理与反射实例演示
  • Hadoop V2 RPC框架使用实例

　　下面开始今天的博客介绍。

2.Java NIO简述

　　Java NIO又称Java New IO，它替代了Java IO API，提供了与标准IO不同的IO工作方式。Java NIO由一下核心组件组成：

  • Channels：连接通道，即能从通道读取数据，又能写数据到通道。可以异步读写，读写从Buffer开始。
  • Buffers：消息缓冲区，用于和NIO通道进行交互。所谓缓冲区，它是一块可以读写的内存，该内存被封装成NIO的Buffer对象，并提供相应的方法，以便于访问。
  • Selectors：通道管理器，它能检测到Java NIO中多个通道，单独的线程可以管理多个通道，间接的管理多个网络连接。

　　下图为Java NIO的工作原理图，如下图所示：

3.Java NIO实例演示

  • NIOServer

　　首先，我们来看NIOServer的代码块。代码内容如下所示：

复制代码 
package cn.hadoop.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.hadoop.conf.ConfigureAPI;

/**
 * @Date May 8, 2015
 *
 * @Author dengjie
 *
 * @Note Defined nio server
 */
public class NIOServer {

    private static final Logger LOGGER = LoggerFactory.getLogger(NIOServer.class);

    // The channel manager
    private Selector selector;

    /**
     * Get ServerSocket channel and initialize
     * 
     * 1.Get a ServerSocket channel
     * 
     * 2.Set channel for non blocking
     * 
     * 3.The channel corresponding to the ServerSocket binding to port port
     * 
     * 4.Get a channel manager
     * 
     * 5.The channel manager and the channel binding, and the channel registered
     * SelectionKey.OP_ACCEPT event
     * 
     * @param port
     * @throws IOException
     */
    public void init(int port) throws IOException {
        ServerSocketChannel serverChannel = ServerSocketChannel.open();
        serverChannel.configureBlocking(false);
        serverChannel.socket().bind(new InetSocketAddress(port));
        this.selector = Selector.open();
        serverChannel.register(selector, SelectionKey.OP_ACCEPT);
    }

    /**
     * listen selector
     * 
     * @throws IOException
     */
    public void listen() throws IOException {
        LOGGER.info("Server has start success");
        while (true) {
            selector.select();
            Iterator<SelectionKey> ite = this.selector.selectedKeys().iterator();
            while (ite.hasNext()) {
                SelectionKey key = (SelectionKey) ite.next();
                ite.remove();
                if (key.isAcceptable()) {
                    ServerSocketChannel server = (ServerSocketChannel) key.channel();
                    SocketChannel channel = server.accept();
                    channel.configureBlocking(false);// 非阻塞
                    channel.write(ByteBuffer.wrap(new String("Send test info to client").getBytes()));
                    channel.register(this.selector, SelectionKey.OP_READ);// 设置读的权限
                } else if (key.isReadable()) {
                    read(key);
                }
            }
        }
    }

    /**
     * Deal client send event
     */
    public void read(SelectionKey key) throws IOException {
        SocketChannel channel = (SocketChannel) key.channel();
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        channel.read(buffer);
        byte[] data = buffer.array();
        String info = new String(data).trim();
        LOGGER.info("Server receive info : " + info);
        ByteBuffer outBuffer = ByteBuffer.wrap(info.getBytes());
        channel.write(outBuffer);// 将消息回送给客户端
    }

    public static void main(String[] args) {
        try {
            NIOServer server = new NIOServer();
            server.init(ConfigureAPI.ServerAddress.NIO_PORT);
            server.listen();
        } catch (Exception ex) {
            ex.printStackTrace();
            LOGGER.error("NIOServer main run error,info is " + ex.getMessage());
        }
    }
}
复制代码
  • NIOClient

　　然后，我们在来看NIOClient的代码块，代码具体内容如下所示：

复制代码 
package cn.hadoop.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.hadoop.conf.ConfigureAPI;

/**
 * @Date May 8, 2015
 *
 * @Author dengjie
 *
 * @Note Defined NIO client
 */
public class NIOClient {

    private static final Logger LOGGER = LoggerFactory.getLogger(NIOClient.class);

    private Selector selector;

    /**
     * Get ServerSocket channel and initialize
     */
    public void init(String ip, int port) throws Exception {
        SocketChannel channel = SocketChannel.open();
        channel.configureBlocking(false);
        this.selector = Selector.open();
        channel.connect(new InetSocketAddress(ip, port));
        channel.register(selector, SelectionKey.OP_CONNECT);
    }

    /**
     * listen selector
     */
    public void listen() throws Exception {
        while (true) {
            selector.select();
            Iterator<SelectionKey> ite = this.selector.selectedKeys().iterator();
            while (ite.hasNext()) {
                SelectionKey key = (SelectionKey) ite.next();
                ite.remove();
                if (key.isConnectable()) {
                    SocketChannel channel = (SocketChannel) key.channel();
                    if (channel.isConnectionPending()) {
                        channel.finishConnect();
                    }
                    channel.configureBlocking(false);// 非阻塞

                    channel.write(ByteBuffer.wrap(new String("Send test info to server").getBytes()));
                    channel.register(this.selector, SelectionKey.OP_READ);
                } else if (key.isReadable()) {
                    read(key);
                }

            }

        }
    }

    /**
     * Deal client send event
     */
    public void read(SelectionKey key) throws IOException {
        SocketChannel channel = (SocketChannel) key.channel();
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        channel.read(buffer);
        byte[] data = buffer.array();
        String info = new String(data).trim();
        LOGGER.info("Client receive info : " + info);
        ByteBuffer outBuffer = ByteBuffer.wrap(info.getBytes());
        channel.write(outBuffer);
    }

    public static void main(String[] args) {
        try {
            NIOClient client = new NIOClient();
            client.init(ConfigureAPI.ServerAddress.NIO_IP, ConfigureAPI.ServerAddress.NIO_PORT);
            client.listen();
        } catch (Exception ex) {
            ex.printStackTrace();
            LOGGER.error("NIOClient main run has error,info is " + ex.getMessage());
        }
    }
}
复制代码
  • ConfigureAPI

　　下面给出ConfigureAPI类的代码，内容如下所示：

复制代码 
package cn.hadoop.conf;

/**
 * @Date May 7, 2015
 *
 * @Author dengjie
 *
 * @Note Defined rpc info
 */
public class ConfigureAPI {

    public interface VersionID {
        public static final long RPC_VERSION = 7788L;
    }

    public interface ServerAddress {
        public static final int NIO_PORT = 8888;
        public static final String NIO_IP = "127.0.0.1";
    }

}
复制代码

4.动态代理和反射简述

　　在Java中，动态代理主要用来做方法的增强，可以在不修改源码的情况下，增强一些方法。另外，还有一个作用就是做远程调用，比如现在有Java接口，该接口的实现部署在非本地服务器上，在编写客户端代码时，由于没法直接生成该对象，这个时候就需要考虑使用动态代理了。

　　而反射，利用了Class类作为反射实例化对象的基本应用，对于一个实例化对象而言，它需要调用类中的构造方法，属性和一般方法，这些操作都可以通过反射机制来完成。下面我们用一个实例来理解这些理论。

5.动态代理和反射实例演示

5.1动态代理

  • JProxy
复制代码 
package cn.java.base;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;

/**
 * @Date May 7, 2015
 * 
 * @Author dengjie
 */
public class JProxy {
    
    public static void main(String[] args) {
        JInvocationHandler ji = new JInvocationHandler();
        Subject sub = (Subject) ji.bind(new RealSubject());
        System.out.println(sub.say("dengjie", 25));
    }

}

interface Subject {
    public String say(String name, int age);
}

class RealSubject implements Subject {

    @Override
    public String say(String name, int age) {
        return name + "," + age;
    }

}

class JInvocationHandler implements InvocationHandler {

    private Object object = null;

    public Object bind(Object object) {
        this.object = object;
        return Proxy.newProxyInstance(object.getClass().getClassLoader(), object.getClass().getInterfaces(), this);
    }

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        Object tmp = method.invoke(this.object, args);
        return tmp;
    }

}
复制代码

5.2反射

  • JReflect
复制代码 
package cn.java.base;

/**
 * @Date May 7, 2015
 * 
 * @Author dengjie
 */
public class JReflect {
    public static void main(String[] args) {
        Fruit f = Factory.getInstance(Orange.class.getName());
        if (f != null) {
            f.eat();
        }
    }
}

interface Fruit {
    public abstract void eat();
}

class Apple implements Fruit {

    @Override
    public void eat() {
        System.out.println("apple");
    }

}

class Orange implements Fruit {

    @Override
    public void eat() {
        System.out.println("orange");
    }

}

class Factory {
    public static Fruit getInstance(String className) {
        Fruit f = null;
        try {
            f = (Fruit) Class.forName(className).newInstance();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return f;
    }
}
复制代码

6.Hadoop V2 RPC框架使用实例

　　本实例主要演示通过Hadoop V2的RPC框架实现一个计算两个整数的Add和Sub，服务接口为 CaculateService ，继承于 VersionedProtocol ，具体代码如下所示：

  • CaculateService
复制代码 
package cn.hadoop.service;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.ipc.ProtocolInfo;
import org.apache.hadoop.ipc.VersionedProtocol;

import cn.hadoop.conf.ConfigureAPI;

/**
 * @Date May 7, 2015
 *
 * @Author dengjie
 *
 * @Note Data calculate service interface
 */
@ProtocolInfo(protocolName = "", protocolVersion = ConfigureAPI.VersionID.RPC_VERSION)
public interface CaculateService extends VersionedProtocol {

    // defined add function
    public IntWritable add(IntWritable arg1, IntWritable arg2);

    // defined sub function
    public IntWritable sub(IntWritable arg1, IntWritable arg2);

}
复制代码

　　注意，本工程使用的是Hadoop-2.6.0版本，这里CaculateService接口需要加入注解，来声明版本号。

　　CaculateServiceImpl类实现CaculateService接口。代码如下所示：

  • CaculateServiceImpl
复制代码 
package cn.hadoop.service.impl;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.ipc.ProtocolSignature;

import cn.hadoop.conf.ConfigureAPI;
import cn.hadoop.service.CaculateService;

/**
 * @Date May 7, 2015
 *
 * @Author dengjie
 *
 * @Note Implements CaculateService class
 */
public class CaculateServiceImpl implements CaculateService {

    public ProtocolSignature getProtocolSignature(String arg0, long arg1, int arg2) throws IOException {
        return this.getProtocolSignature(arg0, arg1, arg2);
    }

    /**
     * Check the corresponding version
     */
    public long getProtocolVersion(String arg0, long arg1) throws IOException {
        return ConfigureAPI.VersionID.RPC_VERSION;
    }

    /**
     * Add nums
     */
    public IntWritable add(IntWritable arg1, IntWritable arg2) {
        return new IntWritable(arg1.get() + arg2.get());
    }

    /**
     * Sub nums
     */
    public IntWritable sub(IntWritable arg1, IntWritable arg2) {
        return new IntWritable(arg1.get() - arg2.get());
    }

}
复制代码

　　CaculateServer服务类，对外提供服务，具体代码如下所示：

  • CaculateServer
复制代码 
package cn.hadoop.rpc;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.ipc.RPC.Server;
import org.slf4j.LoggerFactory;
import org.slf4j.Logger;

import cn.hadoop.service.CaculateService;
import cn.hadoop.service.impl.CaculateServiceImpl;

/**
 * @Date May 7, 2015
 *
 * @Author dengjie
 *
 * @Note Server Main
 */
public class CaculateServer {

    private static final Logger LOGGER = LoggerFactory.getLogger(CaculateServer.class);

    public static final int IPC_PORT = 9090;

    public static void main(String[] args) {
        try {
            Server server = new RPC.Builder(new Configuration()).setProtocol(CaculateService.class)
                    .setBindAddress("127.0.0.1").setPort(IPC_PORT).setInstance(new CaculateServiceImpl()).build();
            server.start();
            LOGGER.info("CaculateServer has started");
            System.in.read();
        } catch (Exception ex) {
            ex.printStackTrace();
            LOGGER.error("CaculateServer server error,message is " + ex.getMessage());
        }
    }

}
复制代码

　　注意，在Hadoop V2版本中，获取RPC下的Server对象不能在使用RPC.getServer()方法了，该方法已被移除，取而代之的是使用Builder方法来构建新的Server对象。

　　RPCClient客户端类，用于访问Server端，具体代码实现如下所示：

  • RPCClient
复制代码 
package cn.hadoop.rpc;

import java.net.InetSocketAddress;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.ipc.RPC;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.hadoop.service.CaculateService;

/**
 * @Date May 7, 2015
 *
 * @Author dengjie
 *
 * @Note RPC Client Main
 */
public class RPCClient {

    private static final Logger LOGGER = LoggerFactory.getLogger(RPCClient.class);

    public static void main(String[] args) {
        InetSocketAddress addr = new InetSocketAddress("127.0.0.1", CaculateServer.IPC_PORT);
        try {
            RPC.getProtocolVersion(CaculateService.class);
            CaculateService service = (CaculateService) RPC.getProxy(CaculateService.class,
                    RPC.getProtocolVersion(CaculateService.class), addr, new Configuration());
            int add = service.add(new IntWritable(2), new IntWritable(3)).get();
            int sub = service.sub(new IntWritable(5), new IntWritable(2)).get();
            LOGGER.info("2+3=" + add);
            LOGGER.info("5-2=" + sub);
        } catch (Exception ex) {
            ex.printStackTrace();
            LOGGER.error("Client has error,info is " + ex.getMessage());
        }
    }

}
复制代码

　　Hadoop V2 RPC服务端截图预览，如下所示：

　　Hadoop V2 RPC客户端截图预览，如下所示：

7.总结

　　Hadoop V2 RPC框架对Socket通信进行了封装，定义了自己的基类接口VersionProtocol。该框架需要通过网络以序列化的方式传输对象，关于Hadoop V2的序列化可以参考《Hadoop2源码分析－序列化篇》，传统序列化对象较大。框架内部实现了基于Hadoop自己的服务端对象和客户端对象。服务端对象通过new RPC.Builder().builder()的方式来获取，客户端对象通过RPC.getProxy()的方式来获取。并且都需要接受Configuration对象，该对象实现了Hadoop相关文件的配置。

8.结束语

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

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

热爱生活，享受编程，与君共勉！



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

文章标签：
Java
Apache
分布式计算
Hadoop
关键词：
hadoop实战
RPC实战
RPC源码分析
hadoop rpc
hadoop源码分析
技术mix呢
目录
相关文章
小小工匠
|
6月前
|
缓存 Java 中间件
jvm性能调优实战 -55RPC调用引发的OOM故障
jvm性能调优实战 -55RPC调用引发的OOM故障
小小工匠
89 0
岁岁种桃花儿
|
6月前
|
Java 应用服务中间件 API
干翻RPC系列之HesssionRPC：HesssionRPC的开发体验和源码分析
干翻RPC系列之HesssionRPC：HesssionRPC的开发体验和源码分析
岁岁种桃花儿
85 0
热烈的马
|
6月前
|
分布式计算 Java 大数据
【大数据技术Hadoop+Spark】HDFS Shell常用命令及HDFS Java API详解及实战（超详细 附源码）
【大数据技术Hadoop+Spark】HDFS Shell常用命令及HDFS Java API详解及实战（超详细 附源码）
热烈的马
699 0
热烈的马
|
6月前
|
分布式计算 大数据 Scala
【大数据技术Hadoop+Spark】Spark RDD创建、操作及词频统计、倒排索引实战（超详细 附源码）
【大数据技术Hadoop+Spark】Spark RDD创建、操作及词频统计、倒排索引实战（超详细 附源码）
热烈的马
291 1
喜欢猪猪
|
5月前
|
存储 分布式计算 Hadoop
Hadoop Distributed File System (HDFS): 概念、功能点及实战
【6月更文挑战第12天】Hadoop Distributed File System (HDFS) 是 Hadoop 生态系统中的核心组件之一。它设计用于在大规模集群环境中存储和管理海量数据，提供高吞吐量的数据访问和容错能力。
喜欢猪猪
636 4
武子康
|
1月前
Hadoop-09-HDFS集群 JavaClient 代码上手实战！详细附代码 安装依赖 上传下载文件 扫描列表 PUT GET 进度条显示（二）
Hadoop-09-HDFS集群 JavaClient 代码上手实战！详细附代码 安装依赖 上传下载文件 扫描列表 PUT GET 进度条显示（二）
武子康
41 3
武子康
|
1月前
|
分布式计算 Java Hadoop
Hadoop-09-HDFS集群 JavaClient 代码上手实战！详细附代码 安装依赖 上传下载文件 扫描列表 PUT GET 进度条显示（一）
Hadoop-09-HDFS集群 JavaClient 代码上手实战！详细附代码 安装依赖 上传下载文件 扫描列表 PUT GET 进度条显示（一）
武子康
40 2
尹正杰
|
2月前
|
分布式计算 Hadoop Devops
Hadoop集群配置https实战案例
本文提供了一个实战案例，详细介绍了如何在Hadoop集群中配置HTTPS，包括生成私钥和证书文件、配置keystore和truststore、修改hdfs-site.xml和ssl-client.xml文件，以及重启Hadoop集群的步骤，并提供了一些常见问题的故障排除方法。
尹正杰
75 3
Hadoop集群配置https实战案例
尹正杰
|
2月前
|
分布式计算 监控 Hadoop
监控Hadoop集群实战篇
介绍了监控Hadoop集群的方法，包括监控Linux服务器、Hadoop指标、使用Ganglia监控Hadoop集群、Hadoop日志记录、通过Hadoop的Web UI进行监控以及其他Hadoop组件的监控，并提供了相关监控工具和资源的推荐阅读链接。
尹正杰
78 2
土木林森
|
3月前
|
资源调度 分布式计算 监控
【揭秘Hadoop YARN背后的奥秘！】从零开始，带你深入了解YARN资源管理框架的核心架构与实战应用！
【8月更文挑战第24天】Hadoop YARN（Yet Another Resource Negotiator）是Hadoop生态系统中的资源管理器，为Hadoop集群上的应用提供统一的资源管理和调度框架。YARN通过ResourceManager、NodeManager和ApplicationMaster三大核心组件实现高效集群资源利用及多框架支持。本文剖析YARN架构及组件工作原理，并通过示例代码展示如何运行简单的MapReduce任务，帮助读者深入了解YARN机制及其在大数据处理中的应用价值。
土木林森
86 0

热门文章

最新文章

  • 1
    Hadoop迁移MaxCompute神器之DataX-On-Hadoop使用指南
  • 2
    Hadoop 2.x本地库native lib及如何源码编译
  • 3
    手把手教你使用自建Hadoop访问全托管服务化HDFS（OSS-HDFS服务）
  • 4
    【centos6.5 hadoop2.7 _64位一键安装脚本】有问题加我Q直接问
  • 5
    linux下安装hadoop步骤
  • 6
    hadoop安装小记
  • 7
    cdh5.3.10_hadoop环境迁移之主机IP修改
  • 8
    发布Apache Hadoop 2.6.0——异构存储，长时间运行的服务与滚动升级支持
  • 9
    Hadoop2.6.0运行mapreduce之推断（speculative）执行（下）
  • 10
    Hadoop 2.x HDFS和YARN的启动方式
  • 1
    实时计算 Flink版产品使用合集之将sdkMode从rpc模式改为jdbc模式后，table.exec.mini-batch.enabled参数还生效吗
    57
  • 2
    java利用xml-rpc协议操作wordpress博客
    84
  • 3
    微服务学习 | Springboot整合Dubbo+Nacos实现RPC调用
    897
  • 4
    一个不错的的rpc开源代码-rest_rpc
    116
  • 5
    【Dubbo3技术专题】「云原生微服务开发实战」 一同探索和分析研究RPC服务的底层原理和实现
    153
  • 6
    【分布式技术专题】「分布式技术架构」实践见真知，手把手教你如何实现一个属于自己的RPC框架（架构技术引导篇）
    267
  • 7
    【Dubbo3技术专题】「服务架构体系」第一章之Dubbo3新特性要点之RPC协议分析介绍
    94
  • 8
    RPC 通信与MQ 消息队列通信的微服务架构区别
    101
  • 9
    Spring Boot整合Dubbo+Zookeeper实现RPC调用
    180
  • 10
    RPC远程服务如何调用
    70

    • 相关课程

    更多
  • 大数据实战项目：反爬虫系统（Lua+Spark+Redis+Hadoop框架搭建）第四阶段
  • 大数据实战项目 - 反爬虫系统（Lua+Spark+Redis+Hadoop框架搭建）第六阶段
  • 大数据实战项目 - 反爬虫系统（Lua+Spark+Redis+Hadoop框架搭建）第七阶段
  • 大数据Hadoop快速入门
  • Hadoop快速入门
  • Hadoop企业优化及扩展案例

    • 相关电子书

    更多
  • 《构建Hadoop生态批流一体的实时数仓》
  • 零基础实现hadoop 迁移 MaxCompute 之 数据
  • CIO 指南:如何在SAP软件架构中使用Hadoop

    • 相关实验场景

    更多
  • 搭建Hadoop环境
    • 下一篇
    阿里云OSS设置跨域访问