文档章节

Hadoop的砖块们--第1章 远程过程调用RPC

brian_2017
 brian_2017
发布于 2017/01/17 09:23
字数 2528
阅读 0
收藏 0

    砖块盖起大厦。“Hadoop的砖块们”,就是逐一分析Hadoop技术的重要技术组成元素。Hadoop的最重要的砖块是远程过程调用RPC。对于RPC来说,《Hadoop技术内幕》一书讲的非常清晰,这里就不重复了。


要先熟悉代理类模式,客户端/服务端网络编程。

1. 代理类

这里的代码主要来自《Hadoop技术内幕》。

packagecom.brian.hadoop.rpc;

importjava.lang.reflect.Proxy;

importjava.lang.reflect.Method;

importjava.lang.reflect.InvocationHandler;

/*

这个例子演示如何使用Proxy代理类

*/

public class MyProxy{

static interfaceCalculatorProtecal{

public int add(int x,int y);

public intsubtract(int x, int y);

}

static class Serverimplements CalculatorProtecal{

public int add(int x,int y){

return x + y;

}

public intsubtract(int x, int y){

return x - y;

}

}

static classCalculatorHandler implements InvocationHandler{

private Objectobjorg;

publicCalculatorHandler(Object obj){

this.objorg =obj;

}

public Objectinvoke(Object Proxy, Method method, Object[] args)

throwsThrowable{

Object result;

System.out.println("\nbefore invoke");

result =method.invoke(this.objorg, args);

System.out.println("\nafter invoke");

return result;

}

}

public static voidmain(String[] args){

System.out.println("ProxyDemo:");

CalculatorProtecalserver = new Server();

InvocationHandlerhandler = new CalculatorHandler(server);

CalculatorProtecalclient = (CalculatorProtecal)Proxy.newProxyInstance(

server.getClass().getClassLoader(),

server.getClass().getInterfaces(),

handler);

int r =client.add(5,1);

int x =client.subtract(5,1);

System.out.println("r= " + r + ", x = " + x);

}

}





2.常规的BS模式

先写一个常规的BS模式,熟悉网络编程的模式。Hadoop用的是NIO,比常规方式更复杂,不太容易理解。

2.1服务端代码Server.java

packagecom.brian.hadoop.net;

importjava.net.ServerSocket;

import java.net.Socket;

import java.io.InputStream;

import java.io.OutputStream;

import java.io.IOException;

import java.io.PrintWriter;

import java.util.Scanner;

import java.lang.Runnable;

import java.lang.Thread;

class Server{

public static voidmain(String[] args){

try{

ServerSocket s =new ServerSocket(8189);

System.out.println("newServersocket ok");

while(true){

Socketincoming = s.accept();

Runnable r =new MyThreadHandler(incoming);

Thread t =new Thread(r);

t.start();

}

}

catch (IOExceptione){

System.out.println("ServerSocketIOException ok");

System.out.println(e);

}

}

}




2.2客户端代码Client.java


package com.brian.hadoop.net;

import java.net.Socket;

importjava.io.DataInputStream;

importjava.io.DataOutputStream;

import java.io.IOException;

import java.io.PrintWriter;

importjava.io.BufferedReader;

importjava.io.InputStreamReader;

import java.lang.Thread;

import java.util.Scanner;

class Client{

public static voidmain(String[] args){

try{

Socket s = newSocket("127.0.0.1", 8189);

System.out.println("newsocket ok");

try{

BufferedReaderin = new BufferedReader(newInputStreamReader(s.getInputStream()));

PrintWriterout = new PrintWriter(s.getOutputStream());

String tmp;

int i = 0;

while (true){

out.println("s--"+ i);

out.flush();

tmp =in.readLine();

System.out.println(tmp);

i++;

}

}finally{

System.out.println("SocketException, finally");

s.close();

}

}

catch (IOExceptione){

System.out.println("SocketIOException ok");

System.out.println(e);

}

}

}



2.3 服务端事件处理的线程类MyThreadHandler

packagecom.brian.hadoop.net;

import java.net.Socket;

import java.lang.Runnable;

import java.io.InputStream;

import java.io.OutputStream;

import java.io.PrintWriter;

import java.io.IOException;

import java.util.Scanner;

class MyThreadHandlerimplements Runnable{

private Socket incoming;

publicMyThreadHandler(Socket incoming){

this.incoming =incoming;

}

public void run(){

try{

InputStreaminstream = this.incoming.getInputStream();

OutputStreamoutstream = this.incoming.getOutputStream();

Scanner in = newScanner(instream);

PrintWriter out =new PrintWriter(outstream, true);

while(in.hasNextLine()){

String line =in.nextLine();

out.println("Echo:"+ line);

if(line.trim().equals("bye")){

break;

}

Thread.sleep(1000);

}

this.incoming.close();

}catch (IOExceptione){

System.out.println(e);

}catch(InterruptedException e){

System.out.println(e);

}

}

}



3. NIO模式

JavaNIO的网络编程比常规的BS模式更快,它是非阻塞的。比较复杂,推荐两个地方:其一,http://xm-king.iteye.com/blog/766330;其二,《JavaNIO》一书。








4. HadoopRPC

RPC就是远程过程调用RemoteProcedure Call。如果本地主机的一个进程,想调用远程主机上的一个进程的某个功能,它应该怎么做呢?

a)远程主机运行服务器端,本地主机运行客户端,通常情况下,远程主机的服务端是一直在运行的。

b)本地主机的客户端通过网络连接到远程主机的指定端口,然后,将要调用的函数名和调用参数通过传给远程主机的服务端。

c)远程主机接收到了函数名和调用参数的时候,它根据函数名找到这个函数,然后根据调用参数执行这个函数,然后把结果通过网络返回到本地主机的客户端。

这里的服务端和客户端的通信方式,可以是常规的B/S模式,也可以是NIOB/S模式,这就是前面的两个例子。


最简单的RPC,可以通过传递字符串的方式进行,也就是说,客户端把要调用的函数名和调用参数,都转换成字符串,然后将字符串传递给服务端,然后在服务端将字符串解析出来,然后调用相应的函数,计算,再讲结果以字符串的方式返回给本地主机客户端,客户端解析字符串获取返回值。这种方式适应于调用参数的类型是简单类型,诸如int,float, char等等。


如果调用参数是对象,字符串方式没效率也不够方便,那么要将参数存储到一个更大的对象里,然后它进行序列化,序列化之后,再从网络传递到远程主机的服务器端,由服务端进行解析,然后调用相应函数,计算,然后将结果序列化,再返回给本地主机的客户端,本地主机的客户端再进行反序列化,获取计算结果。因为远程过程调用多了一个将参数和返回值进行序列化的过程,所以,以代理模式进行调用,代理类在真正进行调用之前对参数进行处理,包括存储和序列化。


4.1 RPC的创建过程

下面,我们以NameNode类和DataNode类为例,看它们是如何实现RPC---- JobTrackerTaskTracker的代码比较长,看起来比较麻烦,就不选它们了。

4.1.1NameNode节点启动

NameNode节点先启动

NameNode.java里,有两个跟RPC有关的import:

importorg.apache.hadoop.ipc.RPC;

importorg.apache.hadoop.ipc.RPC.Server;


NameNode类,定义列RPCServer:

privateServer server;


对象server的由RPC类的getServer函数创建:

this.server= RPC.getServer(this, socAddr.getHostName(), socAddr.getPort(),

handlerCount,false, conf, namesystem.getDelegationTokenSecretManager());


RPC类的getServer函数执行:

newServer(instance, conf, bindAddress, port, numHandlers, verbose,secretManager);

这里的ServerRPC的内部Server类,它继承org.apache.hadoop.ipc.Server类,它的构造函数执行语句是:

super(bindAddress,port, Invocation.class, numHandlers, conf,

classNameBase(instance.getClass().getName()),secretManager);

this.instance= instance;

this.verbose= verbose;


先执行的是org.apache.hadoop.ipc.Server类的构造函数,在它的构造函数,除了设置参数之外,最重要的是创建两个对象:

listener= new Listener();

responder= new Responder();

Listener类是线程类,监听端口。Responder类也是线程类,将RPC结果返回给客户端。


然后,NameNode节点运行org.apache.hadoop.ipc.Serverjoin函数,进入无限循环,运行wait函数。

至此,NameNode节点启动完毕。

4.1.2DataNode节点的启动

DataNode节点的启动晚于NameNode节点的启动。


DataNode节点,引用如下:

importorg.apache.hadoop.ipc.RPC;

importorg.apache.hadoop.ipc.RemoteException;

importorg.apache.hadoop.ipc.Server;


DataNode类定义了一个RPC.Server:

publicServer ipcServer;


ipcServer的创建如下:

ipcServer= RPC.getServer(this, ipcAddr.getHostName(), ipcAddr.getPort(),

conf.getInt("dfs.datanode.handler.count",3), false, conf,

blockTokenSecretManager);

ipcServer的创建过程跟NameNode是相似的,不再详述。

4.1.3DataNode节点和NameNode节点的RPC通信

DataNode节点来说,如何跟NameNode节点进行通信是最重要的,这个通信是RPC方式进行的。DataNode的初始化是在函数startDataNode里进行的。


DataNode节点和NameNode节点以接口DatanodeProtocol进行通信的。


DataNode类,定义如下:

publicDatanodeProtocol namenode = null;

startDataNode函数,创建namenode

this.namenode= (DatanodeProtocol)

RPC.waitForProxy(DatanodeProtocol.class,DatanodeProtocol.versionID, nameNodeAddr, conf);

这里是从远程主机,也就是NameNode节点,创建代理类的对象。


RPCwaitForProxy函数执行,再次调用:

waitForProxy(protocol,clientVersion, addr, conf, 0, Long.MAX_VALUE)

然后,再次调用:

getProxy(protocol,clientVersion, addr, conf, rpcTimeout)

然后,再次调用:

getProxy(protocol,clientVersion, addr, conf, NetUtils.getDefaultSocketFactory(conf),rpcTimeout)

然后,再次调用:

getProxy(protocol,clientVersion, addr, ugi, conf, factory, rpcTimeout)

然后,再次调用:

getProxy(protocol,clientVersion, addr, ticket, conf, factory, rpcTimeout, null)


getProxy函数,创建代理类的关键代码是:

finalInvoker invoker = new Invoker(protocol, addr, ticket, conf, factory,rpcTimeout, connectionRetryPolicy);

VersionedProtocolproxy = (VersionedProtocol)Proxy.newProxyInstance(

protocol.getClassLoader(),new Class[]{protocol}, invoker);


Invoker类是RPC类的内部类,它实现了InvocationHandler接口。Invoker类的构造函数执行如下:

this.remoteId= Client.ConnectionId.getConnectionId(address, protocol,

ticket,rpcTimeout, connectionRetryPolicy, conf);

this.client= CLIENTS.getClient(conf, factory);

这里的Client就是ipc.Client类,ConnectionIdClient的内部类,第一条语句获取一个remoteId,主要是记录各种参数。

CLIENTS的类型是ClientCache类,主要是缓存Client,以供复用,实例化如下:

ClientCacheCLIENTS=new ClientCache();

CLIENTSgetClient函数,会创建client,代码如下:

client= new Client(ObjectWritable.class, conf, factory);

这个创建的过程,也大多是保存各种参数。


对于Proxy.newProxyInstance函数,我们在实现代理类的时候已经熟悉了。

至此,创建代理类的过程就完成了。

4.2 RPC的执行过程

DataNode执行DatanodeProtocal接口的sendHeartBeat函数,过程是怎样的?

DataNode类在offerService函数执行sendHeartbeat函数。


然后,调用RPC的内部类Invoker类的invoke函数,也就是:

invoke(Objectproxy, Method method, Object[] args)


然后,在invoke函数,调用:

ObjectWritablevalue = (ObjectWritable)client.call(new Invocation(method, args),remoteId);


这里的Invocation类是RPC类的内部类。这个类的作用,就是将调用的方法和参数存储在Invocation的实例里,Invocation是可以序列化的,这样就能将调用函数名和参数传递给RPC的服务端。


然后,执行到Client对象的call函数即call(Writableparam, ConnectionId remoteId)

Callcall = new Call(param);

Connectionconnection = getConnection(remoteId, call);

connection.sendParam(call);

synchronized(call) {

while(!call.done) {

...

call.wait();

...

}

}

return call.value;


Call类是Client类的内部类,存贮Invocation类的param,等待执行完了,存储结果。


Connection类是一个线程类,存储call和远程主机的信息,它的sendParam函数把call写入到一个输出流,这个输出流通过Socket方式连接NameNode节点。


然后,数据通过javaNIO通信方式,传递到NameNode的指定端口,这个端口,由Server类的Listener线程类监听。


于是,就到了Server类的Listener线程类。Listener线程类负责接收DataNode节点发过来的数据,然后用Server类的内部类ConnectionReadAndProcess函数进行处理,然后,调用ConnectionprocessData函数。


processData函数会将参数从输入流中读出来,关键代码如下:

Writableparam = ReflectionUtils.newInstance(paramClass, conf);

param.readFields(dis);

Callcall = new Call(id, param, this);

callQueue.put(call);


这里的param就是DataNode节点传的参数param,两者是一样的,它用反序列化的方式从输入流中读出来,然后放到队列callQueue等待执行。


Server类里有一个线程数组,Server启动后,这些线程会运行,这些线程不断地从callQueue取出Call类的对象,执行Call,执行成功后,设置Response,然后由responder线程调用doRespond,把结果回传给DataNode,关键代码如下:

finalCall call = callQueue.take();

...

value= call(call.connection.protocol, call.param, call.timestamp);

...

setupResponse(buf,call, (error == null) ? Status.SUCCESS : Status.ERROR, value,errorClass, error);

...

responder.doRespond(call);


函数call是在在org.apache.hadoop.Server类定义的,但没实现,而是在RPC类的内部类Sever实现了,函数原型如下:

publicWritable call(Class<?> protocol, Writable param, longreceivedTime)


函数call根据Call类的实例call,生成Method,然后调用Method函数获取结果,这个Java的反射,其关键代码如下:

Invocationcall = (Invocation)param;

Methodmethod = protocol.getMethod(call.getMethodName(),call.getParameterClasses());

...

Objectvalue = method.invoke(instance, call.getParameters());

...

return newObjectWritable(method.getReturnType(), value);


responderResponder类的实例,Responder是线程类,它的run函数也有做doRespond的步骤。

至此,RPC的执行过程就结束了。

© 著作权归作者所有

brian_2017
粉丝 3
博文 61
码字总数 145216
作品 0
私信 提问
Hadoop实战-中高级部分 之 Hadoop RPC

Hadoop RestFul Hadoop HDFS原理1 Hadoop HDFS原理2 Hadoop作业调优参数调整及原理 Hadoop HA Hadoop MapReduce高级编程 Hadoop IO Hadoop MapReduce工作原理 Hadoop 管理 Hadoop 集群安装 ...

2k10
2015/03/23
63
0
Hadoop-RPC底层实现与解析

版权声明:博文千万条,版权第一条。转载不规范,博主两行泪 https://blog.csdn.net/qq_39521554/article/details/89431128 Hadoop-RPC底层 RPC 是远程过程调用(Remote Procedure Call),即远...

刺客五六柒
04/21
0
0
Hadoop中RPC机制详解之Server端

Hadoop 中 RPC 机制详解之 Client 端 1. Server.Listener RPC Client 端的 RPC 请求发送到 Server 端后, 首先由 Server.Listener 接收 Server.Listener 类继承自 Thread 类, 监听了 OPREAD 和......

wall--e
2016/05/05
137
0
深入理解RabbitMQ消息队列的使用-张明阳-专题视频课程

版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/a2011480169/article/details/83581214 深入理解RabbitMQ消息队列的使用—126人已学习 课程介绍 RabbitMQ使用...

安静的技术控
2018/06/21
0
0
Hadoop中RPC机制简介

RPC(Remote Procedure Call)—远程过程调用协议,它是一种通过网络从远程计算机程序上请求服务,而不需要了解底层网络技术的协议。RPC协议假定某些传输协议的存在,如TCP或UDP,为通信程序之...

wall--e
2016/04/16
468
0

没有更多内容

加载失败,请刷新页面

加载更多

OpenStack 简介和几种安装方式总结

OpenStack :是一个由NASA和Rackspace合作研发并发起的,以Apache许可证授权的自由软件和开放源代码项目。项目目标是提供实施简单、可大规模扩展、丰富、标准统一的云计算管理平台。OpenSta...

小海bug
13分钟前
1
0
DDD(五)

1、引言 之前学习了解了DDD中实体这一概念,那么接下来需要了解的就是值对象、唯一标识。值对象,值就是数字1、2、3,字符串“1”,“2”,“3”,值时对象的特征,对象是一个事物的具体描述...

MrYuZixian
今天
6
0
数据库中间件MyCat

什么是MyCat? 查看官网的介绍是这样说的 一个彻底开源的,面向企业应用开发的大数据库集群 支持事务、ACID、可以替代MySQL的加强版数据库 一个可以视为MySQL集群的企业级数据库,用来替代昂贵...

沉浮_
今天
4
0
解决Mac下VSCode打开zsh乱码

1.乱码问题 iTerm2终端使用Zsh,并且配置Zsh主题,该主题主题需要安装字体来支持箭头效果,在iTerm2中设置这个字体,但是VSCode里这个箭头还是显示乱码。 iTerm2展示如下: VSCode展示如下: 2...

HelloDeveloper
今天
6
0
常用物流快递单号查询接口种类及对接方法

目前快递查询接口有两种方式可以对接,一是和顺丰、圆通、中通、天天、韵达、德邦这些快递公司一一对接接口,二是和快递鸟这样第三方集成接口一次性对接多家常用快递。第一种耗费时间长,但是...

程序的小猿
今天
7
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部