Apache Mina框架实践

07098

1.为什么要用Apache Mina框架

ApacheMina Server 是一个网络通信应用框架,Mina 可以帮助我们快速开发高性能、高扩展性的网络通信应用，Mina 提供了事件驱动、异步（Mina 的异步IO 默认使用的是JAVANIO 作为底层支持）操作的编程模型。

2.ApacheMina框架使用

Mina的执行流程：

      > IoService：这个接口在一个线程上负责套接字的建立，拥有自己的Selector，监听是否有连接被建立。

      > IoProcessor：这个接口在另一个线程上，负责检查是否有数据在通道上读写，也就是说它也拥有自己的Selector，这是与我们使用JAVA NIO 编码时的一个不同之处，通常在JAVA NIO 编码中，我们都是使用一个Selector，也就是不区分IoService与IoProcessor 两个功能接口。另外，IoProcessor 负责调用注册在IoService 上的过滤器，并在过滤器链之后调用IoHandler。

      >IoFilter：这个接口定义一组拦截器，这些拦截器可以包括日志输出、黑名单过滤、数据的编码（write方向）与解码（read 方向）等功能，其中数据的encode 与decode是最为重要的、也是你在使用Mina 时最主要关注的地方。

      >IoHandler：这个接口负责编写业务逻辑，也就是接收、发送数据的地方。

3.ApacheMina框架实例

       该实例是WIFI云管理平台的实际运用，主要实现通过和WIFI保持长连接管理WIFI,贴出部分代码：

      >服务器端UDPDispatcher.java

      public class UDPDispatcher extendsHttpServlet implements ServletContextListener{

      

      private static final int PORT =20000;

        /** 30秒后超时 */

      private static final int IDELTIMEOUT =30;

      private static final longserialVersionUID = -4384594559203991221L;

      @Override

      public voidcontextInitialized(ServletContextEvent sce) {

             NioDatagramAcceptor acceptor = newNioDatagramAcceptor();// 创建一个UDP的接收器

             acceptor.setHandler(newUDPHandler());// 设置接收器的处理程序

             Executor threadPool = newOrderedThreadPoolExecutor(100);// 建立线程池

             acceptor.getFilterChain().addLast("exector",newExecutorFilter(threadPool));

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

             /*******心跳请求设置 begin*********/

             KeepAliveMessageFactoryheartBeatFactory = new KeepAliveMessageFactoryImpl();

             KeepAliveFilter heartBeat = newKeepAliveFilter(heartBeatFactory,

               IdleStatus.BOTH_IDLE,KeepAliveRequestTimeoutHandler.CLOSE);

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

             heartBeat.setRequestTimeoutHandler(newKeepAliveRequestTimeoutHandlerImpl());

       heartBeat.setForwardEvent(true);

       //设置心跳频率

       heartBeat.setRequestInterval(20);

       heartBeat.setRequestTimeout(60);

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

             /*******心跳请求设置 end  *********/

             DatagramSessionConfig dcfg =acceptor.getSessionConfig();// 建立连接的配置文件

             dcfg.setReadBufferSize(4096);// 设置接收最大字节默认2048

             dcfg.setReceiveBufferSize(1024);//设置输入缓冲区的大小

             dcfg.setSendBufferSize(1024);// 设置输出缓冲区的大小

             dcfg.setReuseAddress(true);// 设置每一个非主监听连接的端口可以重用

             dcfg.setIdleTime(IdleStatus.BOTH_IDLE,IDELTIMEOUT);

             try {

                    // 绑定端口

                    acceptor.bind(newInetSocketAddress(UDPConfigUtil.getHost(),PORT));

             } catch (IOException e) {

                    // TODO Auto-generatedcatch block

                    e.printStackTrace();

             }

       }

      @Override

      public voidcontextDestroyed(ServletContextEvent sce) {

             // TODO Auto-generated method stub

       }

}

      >心跳机制处理KeepAliveMessageFactoryImpl.java

      public void messageSent(IoSessionsession, Object message) throws Exception {

      }

      

      @Override

      public void exceptionCaught(IoSessionsession, Throwable cause)

         throws Exception {

        logger.error(cause);

        session.close(true);

      }

      

      @Override

    public voidmessageReceived(IoSessionsession, Object message) throws Exception {

            logger.info("messageReceived");

       if (message instanceof IoBuffer) {

             IoBuffer buffer = (IoBuffer)message;

             UDPRequestServer controller =UDPRequestServer.getInstance();

                   IoBuffer buffer1 =controller.dealRequest(buffer);

                   session.write(buffer1);

       }

         

    }

}

      >业务逻辑处理UDPHandler.java

publicclass UDPHandler extends IoHandlerAdapter {
      
       private final Logger logger =Logger.getLogger(this.getClass());
         
       @Override
       public void messageSent(IoSessionsession, Object message) throws Exception {
       }  
         
       @Override
       public void exceptionCaught(IoSessionsession, Throwable cause)  
          throws Exception {  
         logger.error(cause);
         session.close(true);  
       }  
         
       @Override
    public void messageReceived(IoSessionsession, Object message) throws Exception {
             logger.info("messageReceived");  
        if (message instanceof IoBuffer) {  
              IoBuffer buffer = (IoBuffer) message;
              UDPRequestServer controller = UDPRequestServer.getInstance();
                   IoBuffer buffer1 =controller.dealRequest(buffer);
                   session.write(buffer1);  
        }
         
    }  

}

注意点：

      >心跳机制：

      (1)客户端会定时发送心跳请求（注意定时时间必须小于，服务器端的IDLE监控时间）,同时需要监听心跳反馈，以此来判断是否与服务器丢失连接。对于服务器的心跳请求不给与反馈。

      (2)心跳情况在60秒之内

      (3)客户端和服务端同时都需要设置心跳请求