设为首页 收藏本站
查看: 909|回复: 0

[经验分享] tomcat NioSender

[复制链接]

尚未签到

发表于 2017-1-15 11:24:06 | 显示全部楼层 |阅读模式
  上次说了NioReceiver,这次看看NioSender,NioSender的体系结构比较复杂的,但感觉不用这么复杂,先看下类图:
  
DSC0000.gif
  这里觉得ParallelNioSender和PooledParallelSender之间有一个类可以去掉,如去了PooledParallelSender,PooledSender类设计成接口,这样的话会比较简洁.
  ChannelCoordinator使用的是PooledParallelSender,最后面最重要的是还是NioSender。
  NioSender是个状态机,看注释:

* - NOT_CONNECTED -> connect() -> CONNECTED
* - CONNECTED -> setMessage() -> READY TO WRITE
* - READY_TO_WRITE -> write() -> READY TO WRITE | READY TO READ
* - READY_TO_READ -> read() -> READY_TO_READ | TRANSFER_COMPLETE
* - TRANSFER_COMPLETE -> CONNECTED
  这样可以循环的利用:

public boolean process(SelectionKey key, boolean waitForAck) throws IOException {
int ops = key.readyOps();
key.interestOps(key.interestOps() & ~ops);
//in case disconnect has been called
if ((!isConnected()) && (!connecting)) throw new IOException("Sender has been disconnected, can't selection key.");
if ( !key.isValid() ) throw new IOException("Key is not valid, it must have been cancelled.");
if ( key.isConnectable() ) {
if ( socketChannel.finishConnect() ) {
completeConnect();
if ( current != null ) key.interestOps(key.interestOps() | SelectionKey.OP_WRITE);
return false;
} else  {
//wait for the connection to finish
key.interestOps(key.interestOps() | SelectionKey.OP_CONNECT);
return false;
}//end if
} else if ( key.isWritable() ) {
boolean writecomplete = write(key);
if ( writecomplete ) {
//we are completed, should we read an ack?
if ( waitForAck ) {
//register to read the ack
key.interestOps(key.interestOps() | SelectionKey.OP_READ);
} else {
//if not, we are ready, setMessage will reregister us for another write interest
//do a health check, we have no way of verify a disconnected
//socket since we don't register for OP_READ on waitForAck=false
read(key);//this causes overhead
setRequestCount(getRequestCount()+1);
return true;
}
} else {
//we are not complete, lets write some more
key.interestOps(key.interestOps()|SelectionKey.OP_WRITE);
}//end if
} else if ( key.isReadable() ) {
boolean readcomplete = read(key);
if ( readcomplete ) {
setRequestCount(getRequestCount()+1);
return true;
} else {
key.interestOps(key.interestOps() | SelectionKey.OP_READ);
}//end if
} else {
//unknown state, should never happen
log.warn("Data is in unknown state. readyOps="+ops);
throw new IOException("Data is in unknown state. readyOps="+ops);
}//end if
return false;
}

  首先屏蔽了SelectionKey的interest集合中的ready集合,只有在这一次处理完以后才会恢复,这样的话下一次的数据不会影响到这一次的数,之后就是 SelectionKey的各种状态的处理,其中read()是调用write()后用于接受ack回执的.
  从ChannelCoordinator开始,来看看NioSender的调用的流程:
  ChannelCoordinator.sendMessage()---->ReplicationTransmitter.sendMessage()---->PooledParallelSender.sendMessage()
  ---->ParallelNioSender.sendMessage()---->ParallelNioSender.doLoop()---->NioSender.process()
  可以吧PooledParallelSender看成一个门面,在这里是NioSender的入口,PooledSender里面主要是一个资源池的作用,这样可以合理利用和控制资源。
  看PooledParallelSender的sendMessage()方法:

if ( !connected ) throw new ChannelException("Sender not connected.");
ParallelNioSender sender = (ParallelNioSender)getSender();
if (sender == null) {
ChannelException cx = new ChannelException("Unable to retrieve a data sender, time out error.");
for (int i = 0; i < destination.length; i++) cx.addFaultyMember(destination, new NullPointerException("Unable to retrieve a sender from the sender pool"));
throw cx;
} else {
try {
sender.sendMessage(destination, message);
sender.keepalive();
} catch (ChannelException x) {
sender.disconnect();
throw x;
} finally {
returnSender(sender);
if (!connected) disconnect();
}
}
  这个地方调用了ParallelNioSender的sendMessage()方法后,还调用了keepalive(),这个方法去除NioSender的集合中不在存活状态的NioSender.最后归还NioSender.
  看下ParallelNioSender的sendMessage()方法,主要是先拿到需要传播的NioSender数组,之后一个个的去传送。
  如果NioSender数组在超时的时间限制内没有发送完成的话,就会抛出异常.主要是在doLoop()方法里面执行的,这里如果出现异常的话,还有NioSender重连等机制试图恢复NioSender的功能
  看tomcat的设计,关于资源池的设计,基本上有两个集合,一个是已经使用的集合,一个是未使用的集合,当未使用集合还有资源的话,现在未使用的集合,当没有的话,在小于最大限度的情况下,可以重新new一个,当一个资源使用完毕的话,放入到未使用的集合可以重新利用。有些时候,一个资源是否可以放入未使用集合是有考虑的,比如他的持续的服务时间是否过长,如果过长就不会放在,因为这个资源过老的话,可能利用的资源不能得到释放,最坏的情况是资源被耗尽,如内存泄漏等,这样还不如重新new一个新的,这些都是我们设计资源池需要考虑的

运维网声明 1、欢迎大家加入本站运维交流群:群②:261659950 群⑤:202807635 群⑦870801961 群⑧679858003
2、本站所有主题由该帖子作者发表,该帖子作者与运维网享有帖子相关版权
3、所有作品的著作权均归原作者享有,请您和我们一样尊重他人的著作权等合法权益。如果您对作品感到满意,请购买正版
4、禁止制作、复制、发布和传播具有反动、淫秽、色情、暴力、凶杀等内容的信息,一经发现立即删除。若您因此触犯法律,一切后果自负,我们对此不承担任何责任
5、所有资源均系网友上传或者通过网络收集,我们仅提供一个展示、介绍、观摩学习的平台,我们不对其内容的准确性、可靠性、正当性、安全性、合法性等负责,亦不承担任何法律责任
6、所有作品仅供您个人学习、研究或欣赏,不得用于商业或者其他用途,否则,一切后果均由您自己承担,我们对此不承担任何法律责任
7、如涉及侵犯版权等问题,请您及时通知我们,我们将立即采取措施予以解决
8、联系人Email:admin@iyunv.com 网址:www.yunweiku.com

所有资源均系网友上传或者通过网络收集,我们仅提供一个展示、介绍、观摩学习的平台,我们不对其承担任何法律责任,如涉及侵犯版权等问题,请您及时通知我们,我们将立即处理,联系人Email:kefu@iyunv.com,QQ:1061981298 本贴地址:https://www.yunweiku.com/thread-328748-1-1.html 上篇帖子: Tomcat Comet 下篇帖子: Tomcat 7 配置tomcat-users.xml【原创】
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

扫码加入运维网微信交流群X

扫码加入运维网微信交流群

扫描二维码加入运维网微信交流群,最新一手资源尽在官方微信交流群!快快加入我们吧...

扫描微信二维码查看详情

客服E-mail:kefu@iyunv.com 客服QQ:1061981298


QQ群⑦:运维网交流群⑦ QQ群⑧:运维网交流群⑧ k8s群:运维网kubernetes交流群


提醒:禁止发布任何违反国家法律、法规的言论与图片等内容;本站内容均来自个人观点与网络等信息,非本站认同之观点.


本站大部分资源是网友从网上搜集分享而来,其版权均归原作者及其网站所有,我们尊重他人的合法权益,如有内容侵犯您的合法权益,请及时与我们联系进行核实删除!



合作伙伴: 青云cloud

快速回复 返回顶部 返回列表