mina中的发送延时

由于项目需要,用到了 mina 框架进行 tcp 通讯。我是初次接触 mina,于是从 Hello world 开始学习了 mina 。期间遇到了一个奇怪的发送数据的延迟问题,解决的过程是曲折的,但找出的原因却令我“吐血”(没真的吐……)。不管怎样,还是贴出来一下作反面案例,希望初次学习 mina 的时候能够绕过这个地雷。

 
hello world 演示很简单,分为两部分,server 和 client 。
 
server 在 8800 端口上起 tcp 侦听,将 client 发送来的消息打印到标准输出(System.out)。只有两个类 HelloServer.java 和 HelloHandler.java,代码如下:

public class HelloServer {

public static void main(String[] args) {

IoBuffer.setUseDirectBuffer(false);

IoBuffer.setAllocator(new SimpleBufferAllocator());

IoAcceptor acceptor = new NioSocketAcceptor();

acceptor.getFilterChain().addLast("logger", new LoggingFilter());

acceptor.getFilterChain().addLast("codec", new ProtocolCodecFilter(new TextLineCodecFactory(

Charset.forName("UTF-8"))));

acceptor.getSessionConfig().setReadBufferSize(2048);

acceptor.getSessionConfig().setIdleTime(IdleStatus.BOTH_IDLE, 10);

acceptor.setHandler(new HelloHandler());

try {

acceptor.bind(new InetSocketAddress(8800));

System.out.println("hello server is running!");

catch (IOException e) {

e.printStackTrace();

}

}

}

 

public class HelloHandler extends IoHandlerAdapter {

@Override

public void sessionCreated(IoSession session) throws Exception {

System.out.println("sessiong created ......");

}

@Override

public void sessionOpened(IoSession session) throws Exception {

System.out.println("session opened ......");

}

@Override

public void sessionClosed(IoSession session) throws Exception {

System.out.println("session closed .");

}

@Override

public void messageReceived(IoSession session, Object message)

throws Exception {

String msgText = message.toString();

System.out.println(getNow() + "message received!-- msg:"

+ msgText);

}

private String getNow(){

Date now = new Date();

DateFormat df = new SimpleDateFormat("[yyyy-MM-dd hh:mm:ss] ");

return df.format(now);

}

@Override

public void messageSent(IoSession session, Object message) throws Exception {

String msgText = message.toString();

System.out.println("message sent!-- msg:" + msgText);

}

@Override

public void exceptionCaught(IoSession session, Throwable cause)

throws Exception {

System.out.println("exception occurred!!! -- " + cause.getMessage());

cause.printStackTrace();

session.close(false);

}

}

 
client 通过 8800 端口连接 server ,发送两条字符串消息。只有两个类 HelloClient.java 和 HelloSender.java, 代码如下:

public class HelloClient {

public static void main(String[] args) {

NioSocketConnector connector = new NioSocketConnector();

connector.setConnectTimeoutMillis(1000 * 15);

connector.getFilterChain().addLast("logger", new LoggingFilter());

connector.getFilterChain().addLast(

"codec",

new ProtocolCodecFilter(new TextLineCodecFactory(Charset

.forName("UTF-8"))));

connector.setHandler(new HelloSender(new String[] {

"Hello message 1 !", "Hello 2" }));

try {

ConnectFuture future = connector.connect(new InetSocketAddress(

"127.0.0.1", 8800));

System.out.println("connect ...");

future.awaitUninterruptibly();

System.out.println("connect future awaitUniterruptibly ...");

IoSession session = future.getSession();

System.out.println("get session");

session.getCloseFuture().awaitUninterruptibly();

System.out.println("session close future awaitUniterruptibly ...");

connector.dispose();

catch (Exception e) {

System.out.println("error !!! --" + e.getMessage());

e.printStackTrace();

}

}

}

 

public class HelloSender extends IoHandlerAdapter {

private String[] msgArray;

public HelloSender(String[] msgArray) {

this.msgArray = msgArray;

}

@Override

public void sessionOpened(IoSession session) throws Exception {

SendMessage(session);

session.close(false);

System.out.println("client handler close session ......");

}

private void SendMessage(IoSession session)throws Exception{

for (int i = 0; i < msgArray.length; i++) {

WriteFuture wf = session.write(msgArray[i]);

wf.addListener(new IoFutureListener<IoFuture>(){

public void operationComplete(IoFuture future) {

System.out.println(getNow() + "futrue -- write completed!");

}

});

System.out.println(getNow() + "--write msg " + i);

Thread.sleep(3000);

}

System.out.println(getNow() + "send completed");

}

private String getNow(){

Date now = new Date();

DateFormat df = new SimpleDateFormat("[yyyy-MM-dd hh:mm:ss] ");

return df.format(now);

}

}

 
先启动服务端,再启动客户端。
 
服务端日志输出如下:
sessiong created ......
session opened ......
[2011-07-13 09:59:19] message received!-- msg:Hello message 1 !
[2011-07-13 09:59:19] message received!-- msg:Hello 2
session closed .
 
 
客户端日志输出如下:
connect ...
connect future awaitUniterruptibly ...
get session
[2011-07-13 09:59:13] --write msg 0
[2011-07-13 09:59:16] --write msg 1
[2011-07-13 09:59:19] send completed
client handler close session ......
[2011-07-13 09:59:19] futrue -- write completed!
[2011-07-13 09:59:19] futrue -- write completed!
session close future awaitUniterruptibly ...
 
仔细观察日志输出就能发现奇怪的问题:客户端是前后相隔 3 秒往 session 中写入数据,但是却在最后同时地发送出去,而服务端也是同时收到这两条消息的,而不是前后相隔 3 秒。
 
一、关于 write 和 flush 的猜想
这也就是说,session 在 write 完成之后并没有立即发送。初步猜测,write 操作时有可能数据只是被放置到缓冲区而已,只有执行 flush 一类的操作才会真正发送出去。
但是 IoSession 类型并没有找到类似 flush 的方法。通过查看源码及 API ,发现 write 的时候,主要逻辑如下:

// Now, we can write the message. First, create a future

WriteFuture writeFuture = new DefaultWriteFuture(this);

WriteRequest writeRequest = new DefaultWriteRequest(message, writeFuture, remoteAddress);

// Then, get the chain and inject the WriteRequest into it

IoFilterChain filterChain = getFilterChain();

filterChain.fireFilterWrite(writeRequest);

每一次写入,都产生一个 WriteRequest 实例,并将该实例传递给 IoSession 的过滤链 filterChain 进行处理。
先说说 IoFilterChain ,其数据结构一个保持 IoFilter 的双向链表。 IoFilterChain 包含两个特殊的 filter : head 和 tail ,用户加入的 filter 都放置于head 和 tail 之间。filterChain.fireFilterWrite(writeRequest) 操作将从 tail 开始往前,逐个地调用 IoFilter 的 filterWrite 方法,最终调用抵达 head 的 filterWrite 方法。想必写入操作就落实在此方法上,进去一探究竟。
 
处于 head 位置的 IoFilter 为 DefaultIoFilter 的内部私有类 HeadFilter 的实例,其主要逻辑如下:

s.getWriteRequestQueue().offer(s, writeRequest);

if (!s.isWriteSuspended()) {

s.getProcessor().flush(s);

}

 
至此,已证实猜测:writeRequest 被放置到一个队列中,当 IoSession 的 WriteSuspended 状态为 false 时,会直接 flush 。但 flush 操作并不是 IoSession 的方法,而是 IoProcessor 的方法。
那就在客户端的 HelloSender 的 SendMessage 方法中语句 WriteFuture wf = session.write(msgArray[i]);   之后加上一句

((AbstractIoSession) session).getProcessor().flush(session);

调试运行 client ,但日志输出和之前一样,两次消息最终还是同时被发送出去的,这招不管用。
跟踪发现 session 的 WriteSuspended 默认就是 false,每次 write 操作都会执行一次 flush,而不需要在 HelloSender 中再次调用 flush 操作。
既然前面补上 flush 操作是多余的,那原因何在?难道 flush 操作中到调用真实的 Channel 进行 send 之间还有缓存?但这与 flush 的语义并不一致。继续查看源码一探究竟。
IoProcessor 的 flush 操作的主要逻辑是把 session 加入到一个 flushingSessions  队列中,然后执行 selector.wakeup 线程执行发送。这 flush 并没有诡异之处,造成先后两次写入的消息同时被发送的原因显然不在于此。
 
二、关于 SocketSessionConfig 的猜想
在如此简单的演示案例中,既然 HelloSender 中的 write 和 flush 操作似乎都没什么问题,难道问题出在 HelloClient 上?
HelloClient 的 main 方法的主要逻辑就以下几步:
1、创建 connector;
2、设置 filterChain;
3、设置 IoHandler 为 HelloSender 的实例;
4、连接服务端,产生会话;
 
这几步操作都是标准的调用,不应该会有问题。但这里面并未对 IoSessionConfig 进行配置,是否是 IoSessionConfig 的默认配置的引起的呢?
于是仔细查看了 IoSessionConfig 的所有属性,并没有属性是与此相关的。
再查看其 IoSessionConfig 派生接口 SocketSessionConfig ,发现有一个 TcpNoDelay 属性。从字面上看,似乎有点关系。
于是在第2步之后加上语句

connector.getSessionConfig().setTcpNoDelay(true);

运行 HelloClient ……

日志输出显示的结果还是一如既往。

NoDelay 是 TCP 层的一个选项,其指示是否将缓存区中的数据合并发送。但在此无论设置为 true 还是 false,都不影响测试的结果。

三、IoHandler 的问题

在进行了以上两方面的尝试后,将怀疑的目光转向了 HelloSender ,这是本例中的 IoHandler 的实现。

发送操作是在 HelloSender 的 sessionOpened 方法执行:

@Override

public void sessionOpened(IoSession session) throws Exception {

SendMessage(session);

session.close(false);

System.out.println("client handler close session ......");

}

当 session 创建后,sessionOpened 便被执行,此时发送数据,并在发送完成后关闭会话的,然后,sessionOpened 方法返回。

仔细想想,似乎在 SendMessage 和 sessionOpened 方法返回之间有些问题:对 sessionOpened 的调用应该是建立连接和会话后初始化过程的一部分,而在初始化过程尚未返回的时候就对 IoSession 写入数据,这也许不是一个恰当的调用方式。也许正是这样,使得先后两次写入的数据都被保持在队列中,直到会话初始化完成后才被处理。在 sessionOpened 方法中创建一个新线程来执行 SendMessage 操作就能验证这一设想,于是 sessionOpened 方法改为如下:

@Override

public void sessionOpened(IoSession session) throws Exception {

final IoSession s = session;

Thread thrd = new Thread(new Runnable() {

public void run() {

try {

SendMessage(s);

Thread.sleep(5000);

s.close(false);

catch (Exception e) {

System.out.println("Send message error!!!--" + e.getMessage());

e.printStackTrace();

}

}

});

thrd.start();

System.out.println("client handler close session ......");

}

 
再次运行测试,日志输出如下:
服务端日志:
hello server is running!
sessiong created ......
session opened ......
[2011-07-14 11:43:01] message received!-- msg:Hello message 1 !
[2011-07-14 11:43:04] message received!-- msg:Hello 2
session closed .
客户端日志:
connect ...
tcpNoDelay=false
connect future awaitUniterruptibly ...
get session
client handler close session ......
[2011-07-14 11:43:01] futrue -- write completed!
[2011-07-14 11:43:01] --write msg 0
[2011-07-14 11:43:04] --write msg 1
[2011-07-14 11:43:04] futrue -- write completed!
[2011-07-14 11:43:07] send completed
session close future awaitUniterruptibly ...
 
日志显示先后间隔 3 秒写入的两条消息在服务端也同样是先后间隔 3 秒被接收到的,一切恢复正常。前面的设想得到证实,原因就是:向 IoSession 写入数据的时机不对。
 
虽然结果是有点雷人,但探索的过程还是收获不少,加深了对 mina 框架的理解。
 
转自:http://www.cnblogs.com/haiq/archive/2011/08/01/2124292.html
上一篇:CentOS安装Git服务器 Centos 6.5 + Git 1.7.1.0 + gitosis


下一篇:foreach加循环体与不加循环体的区别