|
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); } } |
|
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); } } |
|
// 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); |
|
s.getWriteRequestQueue().offer(s, writeRequest); if (!s.isWriteSuspended()) { s.getProcessor().flush(s); } |
((AbstractIoSession) session).getProcessor().flush(session);
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 ......"); } |