mina 心跳机制

http://wandejun1012.iteye.com/blog/2065941

 MINA自带了对心跳协议的支持，可以对心跳做出细致的配置，本文在次基础上实现了server端对client端的心跳检测。

在开始之前先简单介绍下keepAlive的机制：

首先，需要搞清楚TCP keepalive是干什么用的。从名字理解就能够知道，keepalive就是用来检测一个tcp connection是否还连接正常。当一个tcp connection建立好之后，如果双方都不发送数据的话，tcp协议本身是不会发送其它的任何数据的，也就是说，在一个idle的connection上，两个socket之间不产生任何的数据交换。从另一个方面讲，当一个connection建立之后，链接双方可以长时间的不发送任何数据，比如几天，几星期甚至几个月，但该connection仍然存在。

所以，这就可能出现一个问题。举例来说，server和client建立了一个connection，server负责接收client的request。当connection建立好之后，client由于某种原因机器停机了。但server端并不知道，所以server就会一直监听着这个connection，但其实这个connection已经失效了。

keepalive就是为这样的场景准备的。当把一个socket设置成了keepalive，那么这个socket空闲一段时间后，它就会向对方发送数据来确认对方仍然存在。放在上面的例子中，如果client停机了，那么server所发送的keepalive数据就不会有response，这样server就能够确认client完蛋了（至少从表面上看是这样）。

 

具体的源代码如下：

 Server.java

 

 

Java代码  


import java.io.IOException;  

import java.net.InetSocketAddress;  

import java.nio.charset.Charset;  

  

import org.apache.mina.core.service.IoAcceptor;  

import org.apache.mina.core.session.IdleStatus;  

import org.apache.mina.core.session.IoSession;  

import org.apache.mina.filter.codec.ProtocolCodecFilter;  

import org.apache.mina.filter.keepalive.KeepAliveFilter;  

import org.apache.mina.filter.keepalive.KeepAliveMessageFactory;  

import org.apache.mina.filter.logging.LoggingFilter;  

import org.apache.mina.transport.socket.nio.NioSocketAcceptor;  

import org.slf4j.Logger;  

import org.slf4j.LoggerFactory;  

  

public class Server {  

  

    private static final int PORT = 9123;  

    /** 30秒后超时 */  

    private static final int IDELTIMEOUT = 30;  

    /** 15秒发送一次心跳包 */  

    private static final int HEARTBEATRATE = 15;  

    /** 心跳包内容 */  

    private static final String HEARTBEATREQUEST = "0x11";  

    private static final String HEARTBEATRESPONSE = "0x12";  

    private static final Logger LOG = LoggerFactory.getLogger(Server.class);  

  

    public static void main(String[] args) throws IOException {  

        IoAcceptor acceptor = new NioSocketAcceptor();  

        acceptor.getSessionConfig().setReadBufferSize(1024);  

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

                IDELTIMEOUT);  

          

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

        acceptor.getFilterChain().addLast(  

                "codec",  

                new ProtocolCodecFilter(new TextLineCodecFactory()));  

  

        KeepAliveMessageFactory heartBeatFactory = new KeepAliveMessageFactoryImpl();  

        //下面注释掉的是自定义Handler方式  

//      KeepAliveRequestTimeoutHandler heartBeatHandler = new   

//                              KeepAliveRequestTimeoutHandlerImpl();  

//      KeepAliveFilter heartBeat = new KeepAliveFilter(heartBeatFactory,  

//              IdleStatus.BOTH_IDLE, heartBeatHandler);  

          

        KeepAliveFilter heartBeat = new KeepAliveFilter(heartBeatFactory,  

                IdleStatus.BOTH_IDLE);  

  

        //设置是否forward到下一个filter  

        heartBeat.setForwardEvent(true);  

        //设置心跳频率  

        heartBeat.setRequestInterval(HEARTBEATRATE);  

  

        acceptor.getFilterChain().addLast("heartbeat", heartBeat);  

  

        acceptor.setHandler(new MyIoHandler());  

        acceptor.bind(new InetSocketAddress(PORT));  

        System.out.println("Server started on port： " + PORT);  

    }  

  

    /** 

     * @ClassName KeepAliveMessageFactoryImpl 

     * @Description 内部类，实现KeepAliveMessageFactory（心跳工厂） 

     * @author cruise 

     * 

     */  

    private static class KeepAliveMessageFactoryImpl implements  

            KeepAliveMessageFactory {  

  

        @Override  

        public boolean isRequest(IoSession session, Object message) {  

            LOG.info("请求心跳包信息: " + message);  

            if (message.equals(HEARTBEATREQUEST))  

                return true;  

            return false;  

        }  

  

        @Override  

        public boolean isResponse(IoSession session, Object message) {  

//          LOG.info("响应心跳包信息: " + message);  

//          if(message.equals(HEARTBEATRESPONSE))  

//              return true;  

            return false;  

        }  

  

        @Override  

        public Object getRequest(IoSession session) {  

            LOG.info("请求预设信息: " + HEARTBEATREQUEST);  

            /** 返回预设语句 */  

            return HEARTBEATREQUEST;  

        }  

  

        @Override  

        public Object getResponse(IoSession session, Object request) {  

            LOG.info("响应预设信息: " + HEARTBEATRESPONSE);  

            /** 返回预设语句 */  

            return HEARTBEATRESPONSE;  

//          return null;  

        }  

  

    }  

      

    /** 

     * 对应上面的注释 

     * KeepAliveFilter(heartBeatFactory,IdleStatus.BOTH_IDLE,heartBeatHandler) 

     * 心跳超时处理 

     * KeepAliveFilter 在没有收到心跳消息的响应时，会报告给的KeepAliveRequestTimeoutHandler。 

     * 默认的处理是 KeepAliveRequestTimeoutHandler.CLOSE 

     * （即如果不给handler参数，则会使用默认的从而Close这个Session） 

     * @author cruise 

     * 

     */  

  

//  private static class KeepAliveRequestTimeoutHandlerImpl implements  

//          KeepAliveRequestTimeoutHandler {  

//  

//    

//      @Override  

//      public void keepAliveRequestTimedOut(KeepAliveFilter filter,  

//              IoSession session) throws Exception {  

//          Server.LOG.info("心跳超时！");  

//      }  

//  

//  }  

}  

 

MyIoHandler.java

Java代码  


import org.apache.mina.core.service.IoHandlerAdapter;  

import org.apache.mina.core.session.IoSession;  

import org.slf4j.Logger;  

import org.slf4j.LoggerFactory;  

  

  

public class MyIoHandler extends IoHandlerAdapter{  

    private final static Logger log = LoggerFactory  

            .getLogger(MyIoHandler.class);  

  

    @Override  

    public void sessionOpened(IoSession session) throws Exception {  

          

    }  

  

    @Override  

    public void sessionClosed(IoSession session) throws Exception {  

          

    }  

  

    @Override  

    public void messageReceived(IoSession session, Object message)  

            throws Exception {  

        String ip = session.getRemoteAddress().toString();  

        log.info("===> Message From " + ip + " : " + message);         

    }  

      

      

  

}  

  

 

启动Server后，运行telnet客户端连接到Server端，便可以测试心跳；

测试结果如下图：