- 浏览: 13517 次
最新评论
异步机制(Asynchronous) -- (二)异步消息机制兼谈Hadoop RPC
- 博客分类:
- 技术杂绘
异步机制(Asynchronous) -- (二)异步消息机制兼谈Hadoop RPC
2011年01月22日
上篇说了半天,却回避了一个重要的问题:为什么要用异步呢,它有什么样的好处?坦率的说,我对这点的认识不是太深刻(套句俗语,只可意会,不可言传)。还是举个例子吧:
比如Client向Server发送一个request,Server收到后需要100ms的处理时间,为了方便起见,我们忽略掉网络的延迟,并且,我们认为Server端的处理能力是无穷大的。在这个use case下,如果采用同步机制,即Client发送request -> 等待结果 -> 继续发送,那么,一个线程一秒钟之内只能够发送10个request,如果希望达到10000 request/s的发送压力,那么Client端就需要创建1000个线程,而这么多线程的context switch就成为client的负担了。而采用异步机制,就不存在这个问题了。Client将request发送出去后,立即发送下一个request,理论上,它能够达到网卡发送数据的极限。当然,同时需要有机制不断的接收来自Server端的response。
以上的例子其实就是这篇的主题,异步的消息机制,基本的流程是这样的:
如果仔细琢磨的话,会发现这个流程中有两个很重要的问题需要解决:
1. 当client接收到response后,怎样确认它到底是之前哪个request的response呢?
2. 如果发送一个request后,这个request对应的response由于种种原因(比如server端出问题了)一直没有返回。client怎么能够发现类似这样长时间没有收到response的request呢?
对于第一个问题,一般会尝试给每个request分配一个独一无二的ID,返回的Response会同时携带这个ID,这样就能够将request和response对应上了。
对于第二个问题,需要有一个timeout机制,对于每一个request都有一个定时器,如果到指定时间仍然没有返回结果,那么会触发timeout操作。多说一句,timeout机制其实对于涉及网络的同步机制也是非常有必要的,因为有可能client与server之间的链接坏了,在极端情况下,client会被一直阻塞住。
纸上谈兵了这么久,还是看一个实际的例子。我在这里用Hadoop的RPC代码举例。这里需要事先说明的是,Hadoop的RPC对外的接口其实是同步的,但是,RPC的内部实现其实是异步消息机制。多说无益,直接看代码吧(讨论的所有代码都在org.apache.hadoop.ipc.Client.java 里): public Writable call(Writable param, ConnectionId remoteId) throws InterruptedException, IOException { //具体的代码一会再看... } 这就是Client.java对外提供的接口。一共有两个参数,param是希望发送的request,remoteId是指远程server对应的Id。函数的返回就是response(也是继承自writable)。所以说,这是一个同步调用,一旦call函数返回,那么response也就拿到了。
call函数的具体实现一会再看,先介绍Client中两个重要的内部类: private class Call { int id; // call id Writable param; // parameter Writable value; // value, null if error IOException error; // exception, null if value boolean done; // true when call is done protected Call(Writable param) { this.param = param; synchronized (Client.this) { this.id = counter++; } } protected synchronized void callComplete() { this.done = true; notify(); // notify caller } //......... public synchronized void setValue(Writable value) { this.value = value; callComplete(); } public synchronized Writable getValue() { return value; } } call这个类对应的就是一次异步请求。它的几个成员变量:
id: 这个就是之前提过的,对于每一个request都需要分配一个唯一标示符,这样接收到response后才能知道到底对应哪个request;
param: 需要发送到server的request;
value: 从server发送过来的response;
error: 可能发生的异常(比如网络读写错误,server挂了,等等);
done: 表示这个call是否成功完成了,即是否接收到了response; private class Connection extends Thread { private InetSocketAddress server; // server ip:port // ......... private Socket socket = null; // connected socket private DataInputStream in; private DataOutputStream out; //............ // currently active calls private Hashtable calls = new Hashtable(); // ....... private synchronized boolean addCall(Call call) { if (shouldCloseConnection.get()) return false; calls.put(call.id, call); notify(); return true; } private void receiveResponse() { if (shouldCloseConnection.get()) { return; } touch(); try { int id = in.readInt(); // try to read an id if (LOG.isDebugEnabled()) LOG.debug(getName() + " got value #" + id); Call call = calls.get(id); int state = in.readInt(); // read call status if (state == Status.SUCCESS.state) { Writable value = ReflectionUtils.newInstance(valueClass, conf); value.readFields(in); // read value call.setValue(value); calls.remove(id); } else if (state == Status.ERROR.state) { call.setException(new RemoteException(WritableUtils.readString(in), WritableUtils.readString(in))); calls.remove(id); } else if (state == Status.FATAL.state) { // Close the connection markClosed(new RemoteException(WritableUtils.readString(in), WritableUtils.readString(in))); } } catch (IOException e) { markClosed(e); } } public void run() { if (LOG.isDebugEnabled()) LOG.debug(getName() + ": starting, having connections " + connections.size()); try { while (waitForWork()) {//wait here for work - read or close connection receiveResponse(); } } catch (Throwable t) { LOG.warn("Unexpected error reading responses on connection " + this, t); markClosed(new IOException("Error reading responses", t)); } close(); if (LOG.isDebugEnabled()) LOG.debug(getName() + ": stopped, remaining connections " + connections.size()); } public void sendParam(Call call) { if (shouldCloseConnection.get()) { return; } DataOutputBuffer d=null; try { synchronized (this.out) { if (LOG.isDebugEnabled()) LOG.debug(getName() + " sending #" + call.id); //for serializing the //data to be written d = new DataOutputBuffer(); d.writeInt(call.id); call.param.write(d); byte[] data = d.getData(); int dataLength = d.getLength(); out.writeInt(dataLength); //first put the data length out.write(data, 0, dataLength);//write the data out.flush(); } } catch(IOException e) { markClosed(e); } finally { //the buffer is just an in-memory buffer, but it is still polite to // close early IOUtils.closeStream(d); } } } Connection这个类要比之前的Call复杂得多,所以我省略了很多这里不会被讨论的代码。
Connection对应于一个连接,即一个socket。但同时,它又继承自Thread,所有它本身又对应于一个线程。可以看出,在Hadoop的RPC中,一个连接对应于一个线程。先看他的成员变量:
server: 这是远程server的地址;
socket: 对应的socket;
in / out: socket的输入流和输出流;
calls: 重要的成员变量。它是一个hash表, 维护了这个connection正在进行的所有call和它们对应的id之间的关系。当读取到一个response后,就通过id在这张表中找到对应的call;
再看看它的run()函数。这是Connection这个线程的启动函数,我贴的代码中这个函数没做任何的删减,你可以发现,刨除一些冗余代码,这个函数其实就只做了一件事:receiveResponse,即等待接收response。
OK。回到call()这个函数,看看它到底做了什么: public Writable call(Writable param, ConnectionId remoteId) throws InterruptedException, IOException { Call call = new Call(param); Connection connection = getConnection(remoteId, call); connection.sendParam(call); // send the parameter boolean interrupted = false; synchronized (call) { while (!call.done) { try { call.wait(); // wait for the result } catch (InterruptedException ie) { // save the fact that we were interrupted interrupted = true; } } if (interrupted) { // set the interrupt flag now that we are done waiting Thread.currentThread().interrupt(); } if (call.error != null) { if (call.error instanceof RemoteException) { call.error.fillInStackTrace(); throw call.error; } else { // local exception throw wrapException(remoteId.getAddress(), call.error); } } else { return call.value; } } } 首先,它创建了一个新的call(这个call是Call类的实体,注意和call()函数的区分),然后根据remoteId找到对应的connection(Client类中维护了一个connection pool),然后调用connection.sendParam()。从前面找到这个函数,你会发现它就是将request写入到socket,发送出去。
但值得一提的是,它使用的write是最普通的blocking IO,也是同步IO(后面会看到,它读取response也是用的blcoking IO,所以,hadoop RPC虽然是异步机制,但是采用的是同步blocking IO,所以,异步消息机制还采用什么样的IO机制是没有关系的)。
接下来,调用了call.wait(),将线程阻塞在这里。直到在某个地方调用了call.notify(),它才重新运行起来,然后一通判断后返回call.value,即接收到的response。
所以,剩下的问题是,到底是哪调用了call.notify()?
回到connection的receiveResponse函数:
首先,它从socket的输入流中读到一个id,然后根据这个id找到对应的call,调用call.setValue将从socket中读取的response放入到call的value中,然后调用calls.remove(id)将这个call从队列中移除。这里要注意的是call.setValue,这个函数将value设置好之后,调用了call.notify()!
好了,让我们再重头将流程捋一遍:
这里其实有两个线程,一个线程是调用Client.call(),希望向远程server发送请求的线程,另外一个线程就是connection对应的那个线程。当然,虽然有两个线程,但server对应的只有一个socket。第一个线程创建call,然后调用call.sendParam将request通过这个socket发送出去;而第二个线程不断的从socket中读取response。因此,request的发送和response的接收被分隔到不同的线程中执行,而且这两个线程之间关于socket的读写并没有任何的同步机制,因此我认为这个RPC是异步消息机制实现的,只不过通过call.wait()/call.notify()使得对外的接口看上去像是同步。
好了,Hadoop的RPC介绍完了(虽然我略掉了很多内容,比如timeout机制我这里就没写),说说我个人的评价吧。我认为,Hadoop的这个设计还是挺巧妙的,底层采用的是异步机制,但对外的接口提供的又是一般人比较习惯的同步方式。但是,我觉着缺点不是没有,一个问题是一个链接就要产生一个线程,这个如果是在几千台的cluster中,仍然会带来巨大的线程context switch的开销;另一个问题是对于同一个remote server只有一个socket来进行数据的发送和接收,这样的设计网络的吞吐量很有可能上不去。(一家之言,欢迎指正)
未完待续~
发表评论
-
windows 程序设计之「NetTime.C」范例分析笔记
2012-01-20 09:09 722windows 程序设计之「NetTime.C」范例分析笔记 ... -
[How Tomcat Works]第4章 Tomcat默认连接器
2012-01-20 09:09 641[How Tomcat Works]第4章 Tomca ... -
linux服务器编程--EPOLL
2012-01-20 09:09 780linux服务器编程--EPOLL 20 ... -
网络编程常见问题总结
2012-01-20 09:09 966网络编程常见问题总结 2010年10月26日 网络编程常 ... -
VC中的0XCCCCCCCC
2012-01-19 14:13 697VC中的0XCCCCCCCC 2011年10月23日 在 ... -
MySql安装、优化和安全配置(转)
2012-01-19 14:13 647MySql安装、优化和安全配置(转) 2010年11月21日 ... -
嵌入式linux启动信息完全注释(2.2)(转载)
2012-01-19 14:13 569嵌入式linux启动信息完 ... -
推荐一个适合菜鸟使用的小工具
2012-01-19 14:13 472推荐一个适合菜鸟使用的小工具 2011年03月24日 ... -
使用Memcached提高.net应用程序的性能
2012-01-19 14:13 566使用Memcached提高.net应用程序的性能 2012年 ... -
解决系统提示:内存不能为“read”或"written"的办法
2012-01-17 04:03 758解决系统提示:内存不能为“read”或"writte ... -
内存不能为“read”或"written"的解决
2012-01-17 04:03 654内存不能为“read”或"written" ... -
2011年慈溪市小学生计算机程序设计比赛初赛试题
2012-01-17 04:03 9462011年慈溪市小学生计算机程序设计比赛初赛试题 2011年 ... -
C程序1
2012-01-17 04:03 935C程序1 2011年12月15日 ... -
C程序复习题1
2012-01-17 04:03 786C程序复习题1 2011年12月15日 一、正误 ... -
[粤语汉字
2012-01-16 02:51 752[粤语汉字 2010年05月23日 [编辑本段] 粤 ... -
粤语六级
2012-01-16 02:51 741粤语六级 2011年02月26日 粤语六级(节选) ... -
广东话粤语简介
2012-01-16 02:51 584广东话粤语简介 2009年07月19日 粤语 粤语 ... -
粤语发音
2012-01-16 02:51 685粤语发音 2009年08月15日 广州话的声母、韵母和 ...
相关推荐
A-Designer-s-Guide-to-Asynchronous-VLSI
跨平台 Python 异步聊天机器人框架 - Asynchronous multi-platf-nonebot2
xapp523-lvds-4x-asynchronous-oversampling
介绍了异步电机基于matlab/simulink的起动过程分析和仿真,主要包含:1.基于ab0数学模型的异步电动机起动过程分析 2.基于dq0的异步电动机仿真模型及起动过程的仿真
资源来自pypi官网。 资源全名:dm.zodb.asynchronous-1.0.1.tar.gz
with Asynchronous Model Update and TemporallyYang Chen, Xiaoyan Sun, Yaochu Jin,
它将允许您为JSON-RPC或msgpack-RPC构建异步服务器或客户端。 该项目托管在,可在。 文档可在上。概述# include < iostream># include < packio>using packio::arg;using packio::nl_json_rpc::completion_handler;...
swift开发、iOS、macOS、iphone、后台、第2版本
这个程序是基于matlab/simulink的异步电动机直接起动的仿真,我用的软件是matlab2012a
异步上传大文件,用实例介绍异步操作,传输大文件。vs.net工程源码,编译通过,介绍给大家,共同进步。
Grails异步邮件插件 描述 Grails异步邮件是用于异步发送电子邮件的插件。... 存储库软件包(BinTray): https ://bintray.com/kefirsf/plugins/asynchronous-mail/ OpenHUB上的页面: https ://www.openhub.net/p/gra
Sync and Async reset
基于Verilog设计的异步FIFO,有效防止速率不匹配造成的数据丢失,文件中包含源码以及testbench,且有注释,可读性强
可实现异步调用子程序,跨线程访问控件,实现同步调用
将异步电机等效为直流电机,即将定转子解耦的同步旋转模型
asynchronous-parallel-programming:Python中的多线程和多处理异步并行编程
I2C driver asynchronous - I2C EEPROM I2C driver synchronous - I2C EEPROM SPI Driver asynchronous - Self loopback multi client SPI Driver asynchronous - multi instance SPI Driver synchronous - multi ...
异步pika消费者描述此类实现异步RabbitMQ使用者。 它基于并进行了通用修改。 例如,可以使用此修改后的类将消息插入数据库中,或以其他方式处理它们。 Pika是AMQP 0-9-1协议的纯Python实现,包括RabbitMQ的扩展。...
026-asynchronous-native-java-calls-spring
TA-JS-asynchronous-js-TJaaao