mina学习

长连接表示一旦建立了链接,就可以长时间的保持双方的通讯,例如: 
socket链接,推送平台. 
短链接表示建立链接,完成数据的交换之后,就断开链接,例如: http链接.

mina 框架是对socket链接的一次封装框架,可以更好的管理链接的任务. 
在很多的开源项目中使用,例如: Android pn推送框架. 
可以通过简单的几行代码建立通讯链接.

客户端:

        NioSocketConnector connector=new NioSocketConnector();
        //设置处理器
        connector.setHandler(new MyHandler());
        //设置拦截器
        connector.getFilterChain().addLast("codec",
                new ProtocolCodecFilter(new TextLineCodecFactory()));
        //建立连接
        ConnectFuture future=connector.connect(new InetSocketAddress("127.1.1.0", 8888));
        future.awaitUninterruptibly();
        //获取连接的会话
        IoSession session=future.getSession();

        BufferedReader inputReader = new BufferedReader(
                 new InputStreamReader(System.in));
        String inputContent;
        while ((inputContent = inputReader.readLine())!=null) {
            session.write(inputContent);
        }

服务端:

            NioSocketAcceptor  acceptor=new NioSocketAcceptor();
            //设置处理器
            acceptor.setHandler(new MyHandler());
            //设置空闲的时间
            acceptor.getSessionConfig().setIdleTime(
                    IdleStatus.BOTH_IDLE, 3);
            //设置拦截器
            acceptor.getFilterChain().addLast("codec", 
                    new ProtocolCodecFilter(new MyTextLineProtocolFactory()));
            ///端口的绑定
            acceptor.bind(new InetSocketAddress(8888));

至于处理器IohandlerAdapter,与拦截器的编写,略.

---------------------------------------------------

自己以解决：session.getService().getManagedSessions(); 可以当前连接的所用session ；
session.setAttribute(session.getId(), "解析出来的ID") 可以让一个客户上次id 对应一个session；

mina 解决断线重连 ：

http://www.iteye.com/topic/1125178

http://chwshuang.iteye.com/blog/2028951

http://blog.csdn.net/yoara/article/details/37597141

-------------------------

Mina使用IoHandler实现业务处理

IoHandler是Mina实现其业务逻辑的顶级接口；在IoHandler中定义了7个方法，根据I/O事件来触发对应的方法：

import java.io.IOException;

public interface IoHandler {

void sessionCreated(IoSession session) throws Exception;

void sessionOpened(IoSession session) throws Exception;

void sessionClosed(IoSession session) throws Exception;

void sessionIdle(IoSession session, IdleStatus status) throws Exception;

void exceptionCaught(IoSession session, Throwable cause) throws Exception;

void messageReceived(IoSession session, Object message) throws Exception;

void messageSent(IoSession session, Object message) throws Exception;
}
sessionCreated:当一个新的连接建立时，由I/O processor thread调用；

sessionOpened:当连接打开是调用；

messageReceived:当接收了一个消息时调用；

messageSent：当一个消息被(IoSession#write)发送出去后调用；

sessionIdle:当连接进入空闲状态时调用；

sessionClosed:当连接关闭时调用；

exceptionCaught:当实现IoHandler的类抛出异常时调用；

一般情况下，我们最关心的只有messageReceived方法，接收消息并处理，然后调用IoSession的write方法发送出消息！（注意：这里接收到的消息都是Java对象，在IoFilter中所有二进制数据都被解码啦！）

一般情况下很少有人实现IoHandler接口，而是继承它的一个实现类IoHandlerAdapter，这样不用覆盖它的7个方法，只需要根据具体需求覆盖其中的几个方法就可以！

Mina 断线重连

    定义：这里讨论的Mina 断线重连是指使用mina作为客户端软件，连接其他提供Socket通讯服务的服务器端。Socket服务器可以是Mina提供的服务器，也可以是C++提供的服务器。

    
 
 

一、断线重连的方式；

    1. 在创建Mina客户端时增加一个监听器，或者增加一个拦截器，当检测到Session关闭时，自动进行重连。

 
    
 

    2. 在第1种方式的基础上，增加客户端的读写通道空闲检查，当发生Session关闭或者读写空闲时，进行重连。

 
    
 

        第一种方式比较传统，优点是简单方便，适合网络稳定、数据量不大（1M带宽以下）的环境；不过缺点是不能对系统级的连接断开阻塞进行捕获。

       

        第二种方式更加精细，基本上能捕获到应用、网络、系统级的断连。

二、重连目的：

        在使用Mina做为客户端时，往往因为网络、服务器、应用程序出现问题而导致连接断开，而自动重连，就是解决连接断开的唯一方式。如果网线断开、服务器宕机、应用程序挂了，都是断线的原因，这个时候，通过增加一个监听器或者拦截器，就能实现重连。但是生产环境中，断线的原因可能更复杂：网络不稳定、延时、服务器负载高、服务器或者应用程序的发送或者接收缓冲区满等等问题都可能导致数据传输过程出现类似于断线的情况，这个时候，光检测Session关闭是远远不够的，这个时候就需要一种重连机制，比如读写空闲超过30秒，就进行重连。对于数据不间断、实时性高、数据量大的应用场景，更是实用。

三、实例：

    第一种：监听器方式

       创建一个监听器实现mina的IoServiceListener接口，里面的方法可以不用写实现

<span style="font-family: 'Microsoft YaHei',微软雅黑,SimHei,tahoma,arial,helvetica,sans-serif;">import org.apache.mina.core.service.IoService;
import org.apache.mina.core.service.IoServiceListener;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;

public class IoListener implements IoServiceListener{
    @Override
    public void serviceActivated(IoService arg0) throws Exception {
        // TODO Auto-generated method stub
    }
    @Override
    public void serviceDeactivated(IoService arg0) throws Exception {
        // TODO Auto-generated method stub
    }
    @Override
    public void serviceIdle(IoService arg0, IdleStatus arg1) throws Exception {
        // TODO Auto-generated method stub
    }
    @Override
    public void sessionCreated(IoSession arg0) throws Exception {
        // TODO Auto-generated method stub
    }

    @Override
    public void sessionDestroyed(IoSession arg0) throws Exception {
        // TODO Auto-generated method stub
    }
}</span>

     

 再创建客户端时加入监听

<span style="font-family: 'Microsoft YaHei',微软雅黑,SimHei,tahoma,arial,helvetica,sans-serif;">        NioSocketConnector connector = new NioSocketConnector();  //创建连接客户端
        connector.setConnectTimeoutMillis(30000); //设置连接超时
        connector.getSessionConfig().setReceiveBufferSize(10240);   // 设置接收缓冲区的大小
        connector.getSessionConfig().setSendBufferSize(10240);// 设置输出缓冲区的大小
//      加入解码器
        TextLineCodecFactory factory = new TextLineCodecFactory(Charset.forName("GBK"), LineDelimiter.WINDOWS.getValue(), LineDelimiter.WINDOWS.getValue());
        factory.setDecoderMaxLineLength(10240);
        factory.setEncoderMaxLineLength(10240);
        connector.getFilterChain().addLast("codec", new ProtocolCodecFilter(factory));
        connector.setDefaultRemoteAddress(new InetSocketAddress(host, port));// 设置默认访问地址
        //添加处理器
                connector.setHandler(new IoHandler());

                // 添加重连监听
        connector.addListener(new IoListener() {
            @Override
            public void sessionDestroyed(IoSession arg0) throws Exception {
                for (;;) {
                    try {
                        Thread.sleep(3000);
                        ConnectFuture future = connector.connect();
                        future.awaitUninterruptibly();// 等待连接创建成功
                        session = future.getSession();// 获取会话
                        if (session.isConnected()) {
                            logger.info("断线重连[" + connector.getDefaultRemoteAddress().getHostName() + ":" + connector.getDefaultRemoteAddress().getPort() + "]成功");
                            break;
                        }
                    } catch (Exception ex) {
                        logger.info("重连服务器登录失败,3秒再连接一次:" + ex.getMessage());
                    }
                }
            }
        });
                for (;;) {
            try {
                ConnectFuture future = connector.connect();
                future.awaitUninterruptibly(); // 等待连接创建成功
                        session = future.getSession(); // 获取会话
                logger.info("连接服务端" + host + ":" + port + "[成功]" + ",,时间:" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
                break;
            } catch (RuntimeIoException e) {
                logger.error("连接服务端" + host + ":" + port + "失败" + ",,时间:" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()) + ", 连接MSG异常,请检查MSG端口、IP是否正确,MSG服务是否启动,异常内容:" + e.getMessage(), e);
                Thread.sleep(5000);// 连接失败后,重连间隔5s
            }
        }

</span>

    第一种：拦截器方式

<span style="font-family: 'Microsoft YaHei',微软雅黑,SimHei,tahoma,arial,helvetica,sans-serif;">        connector = new NioSocketConnector();  //创建连接客户端
        connector.setConnectTimeoutMillis(30000); //设置连接超时
//      断线重连回调拦截器
        connector.getFilterChain().addFirst("reconnection", new IoFilterAdapter() {
            @Override
            public void sessionClosed(NextFilter nextFilter, IoSession ioSession) throws Exception {
                for(;;){
                    try{
                        Thread.sleep(3000);
                        ConnectFuture future = connector.connect();
                        future.awaitUninterruptibly();// 等待连接创建成功
                        session = future.getSession();// 获取会话
                        if(session.isConnected()){
                            logger.info("断线重连["+ connector.getDefaultRemoteAddress().getHostName() +":"+ connector.getDefaultRemoteAddress().getPort()+"]成功");
                            break;
                        }
                    }catch(Exception ex){
                        logger.info("重连服务器登录失败,3秒再连接一次:" + ex.getMessage());
                    }
                }
            }
        });

        TextLineCodecFactory factory = new TextLineCodecFactory(Charset.forName(encoding), LineDelimiter.WINDOWS.getValue(), LineDelimiter.WINDOWS.getValue());
        factory.setDecoderMaxLineLength(10240);
        factory.setEncoderMaxLineLength(10240);
        //加入解码器
        connector.getFilterChain().addLast("codec", new ProtocolCodecFilter(factory));
        //添加处理器
                connector.setHandler(new IoHandler());
        connector.getSessionConfig().setReceiveBufferSize(10240);   // 设置接收缓冲区的大小
        connector.getSessionConfig().setSendBufferSize(10240);          // 设置输出缓冲区的大小
        connector.setDefaultRemoteAddress(new InetSocketAddress(host, port));// 设置默认访问地址
        for (;;) {
            try {
                ConnectFuture future = connector.connect();
                // 等待连接创建成功
                future.awaitUninterruptibly();
                // 获取会话
                session = future.getSession();
                logger.error("连接服务端" + host + ":" + port + "[成功]" + ",,时间:" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
                break;
            } catch (RuntimeIoException e) {
                logger.error("连接服务端" + host + ":" + port + "失败" + ",,时间:" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()) + ", 连接MSG异常,请检查MSG端口、IP是否正确,MSG服务是否启动,异常内容:" + e.getMessage(), e);
                Thread.sleep(5000);// 连接失败后,重连间隔5s
            }
        }</span>

第二种：加入空闲检测机制

        空闲检测机制需要在创建客户端时，加入空闲超时，然后在处理器handler端的sessionIdle方法中加入一个预关闭连接的方法。让Session关闭传递到监听器或者拦截器的sessionClose方法中实现重连。

      以拦截器方式为例，在创建客户端时，加入读写通道空闲检查超时机制。

<span style="font-family: 'Microsoft YaHei',微软雅黑,SimHei,tahoma,arial,helvetica,sans-serif;">        connector = new NioSocketConnector();  //创建连接客户端
        connector.setConnectTimeoutMillis(30000); //设置连接超时
//      断线重连回调拦截器
        connector.getFilterChain().addFirst("reconnection", new IoFilterAdapter() {
            @Override
            public void sessionClosed(NextFilter nextFilter, IoSession ioSession) throws Exception {
                for(;;){
                    try{
                        Thread.sleep(3000);
                        ConnectFuture future = connector.connect();
                        future.awaitUninterruptibly();// 等待连接创建成功
                        session = future.getSession();// 获取会话
                        if(session.isConnected()){
                            logger.info("断线重连["+ connector.getDefaultRemoteAddress().getHostName() +":"+ connector.getDefaultRemoteAddress().getPort()+"]成功");
                            break;
                        }
                    }catch(Exception ex){
                        logger.info("重连服务器登录失败,3秒再连接一次:" + ex.getMessage());
                    }
                }
            }
        });

        connector.getFilterChain().addLast("mdc", new MdcInjectionFilter());
        TextLineCodecFactory factory = new TextLineCodecFactory(Charset.forName(encoding), LineDelimiter.WINDOWS.getValue(), LineDelimiter.WINDOWS.getValue());
        factory.setDecoderMaxLineLength(10240);
        factory.setEncoderMaxLineLength(10240);
        //加入解码器
        connector.getFilterChain().addLast("codec", new ProtocolCodecFilter(factory));

        connector.getSessionConfig().setReceiveBufferSize(10240);   // 设置接收缓冲区的大小
        connector.getSessionConfig().setSendBufferSize(10240);// 设置输出缓冲区的大小

        connector.getSessionConfig().setIdleTime(IdleStatus.BOTH_IDLE, 30000);  //读写都空闲时间:30秒
        connector.getSessionConfig().setIdleTime(IdleStatus.READER_IDLE, 40000);//读(接收通道)空闲时间:40秒
        connector.getSessionConfig().setIdleTime(IdleStatus.WRITER_IDLE, 50000);//写(发送通道)空闲时间:50秒

        //添加处理器
                connector.setHandler(new IoHandler());

        connector.setDefaultRemoteAddress(new InetSocketAddress(host, port));// 设置默认访问地址
        for (;;) {
            try {
                ConnectFuture future = connector.connect();
                // 等待连接创建成功
                future.awaitUninterruptibly();
                // 获取会话
                session = future.getSession();
                logger.error("连接服务端" + host + ":" + port + "[成功]" + ",,时间:" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
                break;
            } catch (RuntimeIoException e) {
                System.out.println("连接服务端" + host + ":" + port + "失败" + ",,时间:" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()) + ", 连接MSG异常,请检查MSG端口、IP是否正确,MSG服务是否启动,异常内容:" + e.getMessage());
                logger.error("连接服务端" + host + ":" + port + "失败" + ",,时间:" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()) + ", 连接MSG异常,请检查MSG端口、IP是否正确,MSG服务是否启动,异常内容:" + e.getMessage(), e);
                Thread.sleep(5000);// 连接失败后,重连10次,间隔30s
            }
        }</span>

      然后在数据处理器IoHandler中sessionIdle方法中加入Session会话关闭的代码，这样session关闭就能传递到拦截器或者监听器中，然后实现重连。

<span style="font-family: 'Microsoft YaHei',微软雅黑,SimHei,tahoma,arial,helvetica,sans-serif;">import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;

public class IoHandler extends IoHandlerAdapter {
    //部分代码忽略...
    @Override
    public void sessionIdle(IoSession session, IdleStatus status) throws Exception {
        logger.info("-客户端与服务端连接[空闲] - " + status.toString());
        if(session != null){
            session.close(true);
        }
    }
    //部分代码忽略...
}</span>

 总结-最佳实践：

       以上两种方式我个人认为最好是使用第二种。在实际的生产环境，对于数据量比较少的情况下，需要加一个线程专门发送心跳信息，然后在服务器端进行回应心跳，这样就保证读写通道不出现空闲。如果数据量比较大，大到24小时都有数据，那么就不需要心跳线程，可以直接在IoHandler处理器端中messageReceived方法中定时发送心跳到服务器。由于读写监控还可以处理服务器、网络、应用等等方面的不确定因素，所以建议使用第二种方式。