Mina基础(六):Mina整合Spring之前的准备工作——统一通信类、扩展Session及其管理、服务端handler细化、心跳机制及处理

标签: mina扩展  心跳机制  session管理

此章节比较零散,主要为与Spring整合及业务处理做准备,没有涉及到具体的配置,都是一些工具类的实现。

具体的整合Spring,使用的一下的这些类,下面的一章描述了如何使用这些类,看的比较晕的,可以多看看直接的配置,了解mina的运行流程。

完整的项目架构:

统一通信类

  • 规范消息类型

    目的:使用统一的封装类型,服务端接收固定的消息对象,服务端发送固定的消息对象,规范客户端、服务端的交互;

    实现:服务端接收SentBody对象,服务端发送ReplyBody对象

    (参考:spring集成mina 实现消息推送以及转发

Message消息常量:

/**
 * @author ZERO
 * @Description 消息常量
 */
public class Message {
    public static class ReturnCode {
 
        public static String CODE_404 = "404";
 
        public static String CODE_403 = "403";  //该账号未绑定
 
        public static String CODE_405 = "405"; //事物未定义
 
        public static String CODE_200 = "200"; //成功
 
        public static String CODE_500 = "500"; //未知错误
 
    }
 
 
    public static final String SESSION_KEY = "account";
 
 
    /**
     * 服务端心跳请求命令
     */
    public static final String CMD_HEARTBEAT_REQUEST = "hb_request";
 
    /**
     * 客户端心跳响应命令
     */
    public static final String CMD_HEARTBEAT_RESPONSE = "hb_response";
 
    /**
     * 超时次数
     */
    public static final String TIME_OUT_NUM = "timeOutNum";
 
    public static class MessageType {
        // 用户 踢出下线消息类型
        public static String TYPE_999 = "999";
    }
 
}

SentBody服务端接收消息对象:

/**
 * @author ZERO
 * @Description 服务端接收消息对象
 */
public class SentBody implements Serializable {
 
    private static final long serialVersionUID = 1L;
 
    private String key;
 
    private HashMap<String, String> data;
 
    private long timestamp;
 
    public SentBody() {
        data = new HashMap<String, String>();
        timestamp = System.currentTimeMillis();
    }
 
    public String getKey() {
        return key;
    }
 
    public String get(String k) {
        return data.get(k);
    }
 
    public void put(String k, String v) {
        data.put(k, v);
    }
 
    public long getTimestamp() {
        return timestamp;
    }
 
    public void setTimestamp(long timestamp) {
        this.timestamp = timestamp;
    }
 
    public void setKey(String key) {
        this.key = key;
    }
 
    public void remove(String k) {
        data.remove(k);
    }
 
    public HashMap<String, String> getData() {
        return data;
    }
 
    @Override
    public String toString() {
        StringBuffer buffer = new StringBuffer();
        buffer.append("<?xml version=\"1.0\" encoding=\"UTF-8\"?>");
        buffer.append("<sent>");
        buffer.append("<key>").append(key).append("</key>");
        buffer.append("<timestamp>").append(timestamp).append("</timestamp>");
        buffer.append("<data>");
        for (String key : data.keySet()) {
            buffer.append("<" + key + ">").append(data.get(key)).append(
                    "</" + key + ">");
        }
        buffer.append("</data>");
        buffer.append("</sent>");
        return buffer.toString();
    }
 
    public String toXmlString() {
        return toString();
    }
}

ReplyBody服务端发送消息对象:

/**
 * @author ZERO
 * @Description 服务端发送消息对象
 */
public class ReplyBody implements Serializable {

    private static final long serialVersionUID = 1L;

    /**
     * 请求key
     */
    private String key;

    /**
     * 返回码
     */
    private String code;

    /**
     * 返回说明
     */
    private String message;

    /**
     * 返回数据集合
     */
    private HashMap<String, String> data;


    private long timestamp;

    public ReplyBody()
    {
        data = new HashMap<String, String>();
        timestamp = System.currentTimeMillis();
    }
    public long getTimestamp() {
        return timestamp;
    }

    public void setTimestamp(long timestamp) {
        this.timestamp = timestamp;
    }



    public String getKey() {
        return key;
    }

    public void setKey(String key) {
        this.key = key;
    }

    public void put(String k, String v) {
        data.put(k, v);
    }

    public String get(String k) {
        return data.get(k);
    }

    public void remove(String k) {
        data.remove(k);
    }

    public String getMessage() {
        return message;
    }

    public void setMessage(String message) {
        this.message = message;
    }

    public HashMap<String, String> getData() {
        return data;
    }

    public void setData(HashMap<String, String> data) {
        this.data = data;
    }

    public String getCode() {
        return code;
    }

    public void setCode(String code) {
        this.code = code;
    }


    public String toString()
    {

        StringBuilder buffer = new StringBuilder();
        buffer.append("<?xml version=\"1.0\" encoding=\"UTF-8\"?>");
        buffer.append("<reply>");
        buffer.append("<key>").append(this.getKey()).append("</key>");
        buffer.append("<timestamp>").append(timestamp).append("</timestamp>");
        buffer.append("<code>").append(code).append("</code>");
        buffer.append("<message>").append(message).append("</message>");
        buffer.append("<data>");
        for(String key:this.getData().keySet())
        {
            buffer.append("<"+key+">").append(this.get(key)).append("</"+key+">");
        }
        buffer.append("</data>");
        buffer.append("</reply>");
        return buffer.toString();
    }


    public String toXmlString() {
        return toString();
    }

    public String toJson() {
        return new Gson().toJson(this, ReplyBody.class);
    }
}

扩展Session及其管理

目的:方便对session会话进行管理,方便对session会话集合获取和删除

实现:服务端接收到新的Session后,构造一个封装类,实现session 的部分方法,并额外实现方法

Session 封装类:

/**
 * IoSession包装类
 */
public class PcmSession implements Serializable {
 
    // 不参与序列化
    private transient IoSession session;
 
    // 全局ID
    private String gid;
    // session在本机器的ID
    private Long nid;
    // session绑定的服务ip
    private String host;
    // session绑定的账号
    private String account;
    // session绑定的账户消息
    private String message;
    // 扫描数量
    private String scanNum;
    // 经度
    private String longitude;
    // 纬度
    private String latitude;
    // session绑定时间
    private Long bindTime;
    // 心跳时间
    private Long heartbeat;
 
    public PcmSession(){}
 
    public PcmSession(IoSession session) {
        this.session = session;
        this.host = (String) session.getAttribute("address");
        this.nid = session.getId();
    }
 
    /**
     * 将key-value自定义属性,存储到IO会话中
     */
    public void setAttribute(String key, Object value) {
        if (null != session) {
            session.setAttribute(key, value);
        }
    }
 
    /**
     * 从IO的会话中,获取key的value
     */
    public Object getAttribute(String key) {
        if (null != session) {
            return session.getAttribute(key);
        }
        return null;
    }
 
    /**
     *  在IO的会话中,判断是否存在包含key-value
     */
    public boolean containsAttribute(String key) {
        if (null != session) {
            return session.containsAttribute(key);
        }
        return false;
    }
 
    /**
     * 从IO的会话中,删除key
     */
    public void removeAttribute(String key) {
        if (null != session) {
            session.removeAttribute(key);
        }
    }
 
    /**
     *  获取IP地址
     */
    public SocketAddress getRemoteAddress() {
        if (null != session) {
            return session.getRemoteAddress();
        }
        return null;
    }
 
    /**
     * 将消息对象 message发送到当前连接的对等体(异步)
     * 当消息被真正发送到对等体的时候,IoHandler.messageSent(IoSession,Object)会被调用。
     * @param msg 发送的消息
     */
    public void write(Object msg) {
        if (null != session) {
            CustomPack pack = new CustomPack((String) msg);
            session.write(pack).isWritten();
        }
    }
 
    /**
     * 发送消息重载,是否为请求
     * @param msg 发送的消息
     * @param isRequest 是否为请求
     */
    public void write(Object msg, boolean isRequest) {
        if (null != session) {
            byte flag = isRequest ? CustomPack.REQUEST : CustomPack.RESPONSE;
            CustomPack pack = new CustomPack(flag, (String) msg);
            session.write(pack).isWritten();
        }
    }
 
    /**
     * 会话是否已经连接
     */
    public boolean isConnected() {
        if (null != session) {
            return session.isConnected();
        }
        return false;
    }
 
    /**
     *  会话是否为本地连接
     */
    public boolean isLocalHost() {
        try {
            String ip = InetAddress.getLocalHost().getHostAddress();
            return ip.endsWith(host);
        } catch (UnknownHostException e) {
            e.printStackTrace();
        }
        return false;
    }
 
    /**
     * 关闭当前连接。如果参数 immediately为 true的话
     * 连接会等到队列中所有的数据发送请求都完成之后才关闭;否则的话就立即关闭。
     */
    public void close(boolean immediately) {
        if (null != session) {
            if (immediately) {
                session.closeNow();
            } else {
                session.closeOnFlush();
            }
        }
    }
 
    @Override
    public boolean equals(Object obj) {
        if (this == obj) {
            return true;
        }
        if (null == obj) {
            return false;
        }
        if (this.getClass() != obj.getClass()) {
            return false;
        }
        // 强转为当前类
        PcmSession session = (PcmSession) obj;
        if (session.nid != null && nid != null) {
            return session.nid.longValue() == nid.longValue() && session.host.equals(host);
        }
        return false;
    }
 
    @Override
    public String toString() {
        StringBuilder buffer = new StringBuilder();
        buffer.append("{");
        buffer.append("\"").append("gid").append("\":").append("\"").append(gid).append("\"").append(",");
        buffer.append("\"").append("nid").append("\":").append(nid).append(",");
        buffer.append("\"").append("host").append("\":").append("\"").append(host).append("\"").append(",");
        buffer.append("\"").append("account").append("\":").append("\"").append(account).append("\"").append(",");
        buffer.append("\"").append("message").append("\":").append(message).append(",");
        buffer.append("\"").append("longitude").append("\":").append(longitude).append(",");
        buffer.append("\"").append("latitude").append("\":").append(latitude).append(",");
        buffer.append("\"").append("bindTime").append("\":").append(bindTime).append(",");
        buffer.append("\"").append("heartbeat").append("\":").append(heartbeat);
        buffer.append("}");
        return buffer.toString();
    }
 
    public IoSession getSession() {
        return session;
    }
 
    public void setSession(IoSession session) {
        this.session = session;
    }
 
    public String getGid() {
        return gid;
    }
 
    public void setGid(String gid) {
        this.gid = gid;
    }
 
    public Long getNid() {
        return nid;
    }
 
    public void setNid(Long nid) {
        this.nid = nid;
    }
 
    public String getHost() {
        return host;
    }
 
    public void setHost(String host) {
        this.host = host;
    }
 
    public String getAccount() {
        return account;
    }
 
    public void setAccount(String account) {
        this.account = account;
    }
 
    public String getMessage() {
        return message;
    }
 
    public void setMessage(String message) {
        this.message = message;
    }
 
    public String getScanNum() {
        return scanNum;
    }
 
    public void setScanNum(String scanNum) {
        this.scanNum = scanNum;
    }
 
    public String getLongitude() {
        return longitude;
    }
 
    public void setLongitude(String longitude) {
        this.longitude = longitude;
    }
 
    public String getLatitude() {
        return latitude;
    }
 
    public void setLatitude(String latitude) {
        this.latitude = latitude;
    }
 
    public Long getBindTime() {
        return bindTime;
    }
 
    public void setBindTime(Long bindTime) {
        this.bindTime = bindTime;
    }
 
    public Long getHeartbeat() {
        return heartbeat;
    }
 
    public void setHeartbeat(Long heartbeat) {
        this.heartbeat = heartbeat;
    }
 
}

Session管理接口:

/**
 * Session 管理接口
 */
public interface SessionManager {
    /**
     * 添加session
     */
    void addSession(String account, PcmSession session);
 
    /**
     * 获取session
     */
    PcmSession getSession(String account);
 
    /**
     * 替换Session
     */
    void replaceSession(String account, PcmSession session);
 
    /**
     * 删除session
     */
    void removeSession(String account);
 
    /**
     * 删除session
     */
    void removeSession(PcmSession pcmSession);
 
}

    SessionManager接口实现类DefaultSessionManagerImpl

    实现:内部创建一个线程安全的ConcurrentHashMap,存放封装的session对象

/**
 * 默认session管理接口实现类
 */
public class DefaultSessionManagerImpl extends Observable implements SessionManager {
 
    /**
     * 存放session的线程安全的map集合
     */
    private static ConcurrentHashMap<String, PcmSession> sessions = new ConcurrentHashMap<>();
 
    /**
     * 线程安全的自增类,用于统计连接数
     */
    private static final AtomicInteger connectionsCounter = new AtomicInteger(0);
 
 
    /**
     * 添加session
     * @param account
     * @param session
     */
    @Override
    public void addSession(String account, PcmSession session) {
        if (null != session) {
            sessions.put(account, session);
            connectionsCounter.incrementAndGet();
            // 被观察者方法,拉模型
            setChanged();
            notifyObservers();
        }
    }
 
    /**
     * 获取session
     * @param account
     * @return
     */
    @Override
    public PcmSession getSession(String account) {
        return sessions.get(account);
    }
 
    /**
     * 替换session方法,通过account
     */
    @Override
    public void replaceSession(String account, PcmSession session) {
        sessions.put(account, session);
        // 被观察者方法,拉模型
        setChanged();
        notifyObservers();
    }
 
 
    /**
     * 移除session通过account
     * @param account
     */
    @Override
    public void removeSession(String account) {
        sessions.remove(account);
        connectionsCounter.decrementAndGet();
        // 被观察者方法,拉模型
        setChanged();
        notifyObservers();
    }
 
    /**
     * 移除session通过session
     * @param pcmSession
     */
    @Override
    public void removeSession(PcmSession pcmSession) {
        String account = (String) pcmSession.getAttribute(Message.SESSION_KEY);
        removeSession(account);
    }
 
    public static ConcurrentHashMap<String, PcmSession> getSessions() {
        return sessions;
    }
 
 
}

服务端handler细化

    目的:方便对客户端发送过来的数据处理,

    实现:服务端接收到客户端的消息后,根据消息中封装的key,使用对应的key处理方式,实现一个统一的接口

定义处理接口:

/**
 * Mina的请求处理接口,必须实现此接口
 */
public interface RequestHandler {
    ReplyBody process(PcmSession session, SentBody sent);
}

这里我们实现了三个具体的handler是BindHandler、PushMessageHandler、SessionClosedHandler分别代表是绑定、推送、关闭。

BindHandler接口实现:

/**
 * 绑定处理handler
 */
public class BindHandler implements RequestHandler{
 
    private final Logger logger = LogManager.getLogger(BindHandler.class);
 
    /**
     * 逻辑处理方法
     * @param newSession 新的会话
     * @param message 接收的信息
     * @return
     */
    @Override
    public ReplyBody process(PcmSession newSession, SentBody message) {
        ReplyBody reply = new ReplyBody();
        // 获取会话管理类
        SessionManager sessionManager = (DefaultSessionManagerImpl) SpringContextUtil.getBean("pcmSessionManager");
        try {
            String account = message.get(Message.SESSION_KEY);
            newSession.setAccount(account);
            newSession.setAttribute(Message.SESSION_KEY, account);
            newSession.setAttribute(Message.TIME_OUT_NUM, 0); // 超时次数设为0
            newSession.setGid(UuidUtil.get32UUID());
            // 设置部分所需信息
            newSession.setMessage(message.get("message"));
            newSession.setScanNum(message.get("scanNum"));
            newSession.setLongitude(message.get("longitude"));
            newSession.setLatitude(message.get("latitude"));
            // 设置绑定时间,第一次心跳时间
            newSession.setBindTime(System.currentTimeMillis());
            newSession.setHeartbeat(System.currentTimeMillis());
            // 由于客户端断线服务端可能会无法获知的情况,客户端重连时,需要关闭旧的连接
            PcmSession oldSession = sessionManager.getSession(account);
            if (oldSession != null && !oldSession.equals(newSession)) {
                // 移除account属性
                oldSession.removeAttribute(Message.SESSION_KEY);
                // 替换oldSession
                sessionManager.replaceSession(account, newSession);
                // 发送t下线的消息
                ReplyBody rb = new ReplyBody();
                rb.setCode(Message.MessageType.TYPE_999);
                rb.put(Message.SESSION_KEY, account);
                // 判断当前会话是否是属于本地的会话
                if (oldSession.isLocalHost()) {
                    oldSession.write(rb.toJson());
                    oldSession.close(true);
                    logger.info(">>>>>>>>>>>>>>>>>> 终端用户:" + account + "已在别处登陆,当前连接已被关闭 <<<<<<<<<<<<<<<<<" );
                } else {
                    // 不是则需要发往目标服务器处理
                    // 本服务为提供此功能,需要自行添加
                }
            }
            if (oldSession == null) {
                sessionManager.addSession(account, newSession);
            }
            reply.setCode(Message.ReturnCode.CODE_200);
        } catch (Exception e) {
            reply.setCode(Message.ReturnCode.CODE_500);
            e.printStackTrace();
        }
        if (reply.getCode().equals(Message.ReturnCode.CODE_200)) {
            logger.info(">>>>>>>>>>>>>>>>>> 终端用户:" + message.get(Message.SESSION_KEY) + "绑定成功 <<<<<<<<<<<<<<<<<<<<");
        } else {
            logger.info(">>>>>>>>>>>>>>>>>> 终端用户:" + message.get(Message.SESSION_KEY) + "绑定失败 <<<<<<<<<<<<<<<<<<<<");
        }
        return reply;
    }
}

PushMessageHandler实现:

/**
 * 推送消息的handler
 */
public class PushMessageHandler implements RequestHandler{
 
    private final Logger logger = LogManager.getLogger(PushMessageHandler.class);
 
    @Override
    public ReplyBody process(PcmSession session, SentBody sent) {
        ReplyBody reply = new ReplyBody();
        // 获取绑定的账户
        String account = sent.getData().get(Message.SESSION_KEY);
        SessionManager sessionManager = (DefaultSessionManagerImpl)SpringContextUtil.getBean("pcmSessionManager");
        // 获取会话
        PcmSession ios = sessionManager.getSession(account);
        if (ios != null) {
            sent.remove(Message.SESSION_KEY);
            reply.setKey(sent.getKey());
            reply.setMessage("推送的消息");
            reply.setData(sent.getData());
            reply.setCode(Message.ReturnCode.CODE_200);
            ios.write(reply.toJson());
            logger.info(">>>>>>>>>>>>>>>>>> 服务器发送消息成功,接收用户:" + session.getAccount() + " >>>>>>>>>>>>>>>>>>");
        } else {
            reply.setCode(Message.ReturnCode.CODE_500);
            reply.setMessage("Mina push message fail");
        }
        return reply;
    }
}

 

 

SessionClosedHandler实现:

/**
 * 会话关闭处理
 */
public class SessionClosedHandler implements RequestHandler{
 
    /**
     * 逻辑处理方法
     * @param session
     * @param message
     * @return
     */
    @Override
    public ReplyBody process(PcmSession session, SentBody message) {
        ReplyBody rb = new ReplyBody();
        // 获取会话管理类
        SessionManager sessionManager = (DefaultSessionManagerImpl) SpringContextUtil.getBean("pcmSessionManager");
        if (session.getAttribute(Message.SESSION_KEY) == null) {
            return null;
        }
        // 在管理类的map中移除
        String account = session.getAttribute(Message.SESSION_KEY).toString();
        sessionManager.removeSession(account);
        return null;
    }
}

Mina服务端消息handler:

/**
 * 服务端handler
 */
public class ServiceHandler extends IoHandlerAdapter {
 
    private final Logger logger = LogManager.getLogger(ServiceHandler.class);
    // 存放本地处理的handler
    private HashMap<String, RequestHandler> handlers = new HashMap<String, RequestHandler>();
 
    /**
     * 接收到消息时
     * @param session
     * @param message
     * @throws Exception
     */
    @Override
    public void messageReceived(IoSession session, Object message) throws Exception {
        logger.info("<<<<<<<<<<<<<<<<<<<< 获取一条信息,来自sessionId:" + session.getId() + " <<<<<<<<<<<<<<<<<<<<");
        // 转为自定义协议,取出内容,转为接受的对象
        SentBody sentBody = new Gson().fromJson(((CustomPack)message).getContent(), SentBody.class);
        ReplyBody rb = new ReplyBody();
        PcmSession pcmSession = new PcmSession(session);
        String key = sentBody.getKey();
        // 根据key的不同调用不同的handler
        RequestHandler handler = handlers.get(key);
        // 如果没有这个handler
        if (handler == null) {
            rb.setCode(Message.ReturnCode.CODE_405);
            rb.setMessage("Service undefined this handler :" + key);
        } else {
            rb = handler.process(pcmSession, sentBody);
        }
        rb.setKey(key);
        pcmSession.write(rb.toJson(), false);
    }
 
    /**
     * 发送消息
     * @param session
     * @param message
     * @throws Exception
     */
    @Override
    public void messageSent(IoSession session, Object message) throws Exception {
        // session.closeOnFlush(); 需要短连接,则发送后关闭连接
        logger.debug(">>>>>>>>>>>>>>>>>>>> 发送消息成功 >>>>>>>>>>>>>>>>>>>>");
    }
 
    /**
     * 建立连接时
     * @param session
     * @throws Exception
     */
    @Override
    public void sessionCreated(IoSession session) throws Exception {
        InetSocketAddress isa = (InetSocketAddress) session.getRemoteAddress();
        // IP
        String address = isa.getAddress().getHostAddress();
        session.setAttribute("address", address);
        logger.info(">>>>>>>>>>>>>>>>>> 来自" + address + " 的终端上线,sessionId:" + session.getId() + "  <<<<<<<<<<<<<");
    }
 
    /**
     * 打开连接时
     * @param session
     * @throws Exception
     */
    @Override
    public void sessionOpened(IoSession session) throws Exception {
        logger.debug("Open a connection ...");
    }
 
    /**
     * 连接空闲时
     * @param session
     * @param status
     * @throws Exception
     */
    @Override
    public void sessionIdle(IoSession session, IdleStatus status) throws Exception {
        logger.debug("sessionIdle ... from " + session.getRemoteAddress());
    }
 
    /**
     * 关闭连接时
     * @param session
     * @throws Exception
     */
    @Override
    public void sessionClosed(IoSession session) throws Exception {
        PcmSession pcmSession = new PcmSession(session);
        // 获取连接关闭的handler,进行处理
        try {
            RequestHandler handler = handlers.get("clientClose");
            if (handler != null && pcmSession.containsAttribute(Message.SESSION_KEY)) {
                handler.process(pcmSession, null);
                logger.info(">>>>>>>>>>>>>>>>>> 终端用户:" + session.getAttribute(Message.SESSION_KEY) + "已下线 <<<<<<<<<<<<<<<<<" );
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
//        pcmSession.close(true);
    }
 
    /**
     * 捕获到异常
     * @param session
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(IoSession session, Throwable cause) throws Exception {
        logger.error(">>>>>>>>>>>>>>>>>> 终端用户:" + session.getAttribute(Message.SESSION_KEY) + "连接发生异常,即将关闭连接,原因:" + cause.getMessage() + " <<<<<<<<<<<<<<<<<");
    }
 
    public HashMap<String, RequestHandler> getHandlers() {
        return handlers;
    }
 
    public void setHandlers(HashMap<String, RequestHandler> handlers) {
        this.handlers = handlers;
        logger.info(">>>>>>>>>>>>>>>>>> Mina服务端启动成功 <<<<<<<<<<<<<<<<<");
    }
}

 

心跳机制及处理

先简单介绍下keepAlive的机制:
    首先,需要搞清楚TCP keepalive是干什么用的。从名字理解就能够知道,keepalive就是用来检测一个tcp connection是否还连接正常。当一个tcpconnection建立好之后,如果双方都不发送数据的话,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完蛋了(至少从表面上看是这样)。

MINA本身提供了一个过滤器类: org.apache.mina.filter.keepalive . KeepAliveFilter ,该过滤器用于在IO空闲的时候发送并且反馈心跳包(keep-alive request/response)。 
该类构造函数中参数有三个分别是: 
(1)KeepAvlieMessageFactory:   该实例引用用于判断接受与发送的包是否是心跳包,以及心跳请求包的实现 
(2)IdleStatus:   该过滤器所关注的空闲状态,默认认为读取空闲。 即当读取通道空闲的时候发送心跳包 
(3)KeepAliveRequestTimeoutHandler: 心跳包请求后超时无反馈情况下的处理机制  默认为CLOSE  即关闭连接 

参考:
http://www.cnblogs.com/pricks/p/3832882.html
https://blog.csdn.net/kkk0526/article/details/51732437

首先,实现KeepAvlieMessageFactory接口:

/**
 * 心跳实现类
 * 服务端发送的是hb_request,那么客户端就应该返回hb_response
 */
public class KeepAliveFactoryImpl implements KeepAliveMessageFactory {
 
    private final Logger logger = LogManager.getLogger(KeepAliveFactoryImpl.class);
 
    // 服务端需要发送请求,客户端无需发送
    private boolean sendHbRequest;
 
    public KeepAliveFactoryImpl(boolean isServer) {
        this.sendHbRequest = isServer;
    }
 
    /**
     * 服务端心跳发送请求命令
     */
    private static final String HEART_BEAT_REQUEST = Message.CMD_HEARTBEAT_REQUEST;
 
    /**
     * 客户端心跳响应命令
     */
    private static final String HEART_BEAT_RESPONSE = Message.CMD_HEARTBEAT_RESPONSE;
 
    // 在需要发送心跳时,用来获取一个心跳请求包[发送端使用]
    @Override
    public Object getRequest(IoSession session) {
        // 是否需要发送心跳请求
        if (sendHbRequest) {
            return new CustomPack(HEART_BEAT_REQUEST);
        }
        return null;
    }
 
    // 在需要回复心跳时,用来获取一个心跳回复包[接收端使用]
    @Override
    public Object getResponse(IoSession session, Object request) {
        return new CustomPack(CustomPack.RESPONSE, HEART_BEAT_RESPONSE);
    }
 
    // 用来判断接收到的消息是不是一个心跳请求包,是就返回true[接收端使用]
    @Override
    public boolean isRequest(IoSession session, Object message) {
        if (message instanceof CustomPack) {
            CustomPack pack = (CustomPack) message;
            if (pack.getContent().equals(Message.CMD_HEARTBEAT_REQUEST)) {
                // 将超时次数置为0
                session.setAttribute(Message.TIME_OUT_NUM, 0);
                return true;
            }
            return false;
        }
        return false;
    }
 
    // 用来判断接收到的消息是不是一个心跳回复包,是就返回true[发送端使用]
    @Override
    public boolean isResponse(IoSession session, Object message) {
        if (message instanceof CustomPack) {
            CustomPack pack = (CustomPack) message;
            return pack.getContent().equals(Message.CMD_HEARTBEAT_RESPONSE);
        }
        return false;
    }
 
}

之后实现超时处理类实现KeepAliveRequestTimeoutHandler接口,业务属性为:超时3次后,关闭连接:

/**
 * 心跳超时处理类
 */
public class KeepAliveRequestTimeoutHandlerImpl implements KeepAliveRequestTimeoutHandler {
 
    private static final Logger logger = LogManager.getLogger(KeepAliveRequestTimeoutHandlerImpl.class);
 
    /**
     * 超时的最大次数
     */
    private int timeoutNum = 3;
 
    public KeepAliveRequestTimeoutHandlerImpl() {}
 
    public KeepAliveRequestTimeoutHandlerImpl(int timeoutNum) {
        this.timeoutNum = timeoutNum;
    }
 
    @Override
    public void keepAliveRequestTimedOut(KeepAliveFilter filter, IoSession session) throws Exception {
        int isTimeoutNum = (int) session.getAttribute(Message.TIME_OUT_NUM);
        // 没有超过最大次数,超时次数加1
        if (isTimeoutNum <= timeoutNum) {
            session.setAttribute(Message.TIME_OUT_NUM, isTimeoutNum + 1);
        } else {
            // 超过最大次数,关闭会话连接
            SessionManager sessionManager = (SessionManager) SpringContextUtil.getBean("pcmSessionManager");
            String account = (String) session.getAttribute(Message.SESSION_KEY);
            sessionManager.removeSession(account);
            logger.info("<<<<<<<<<<<<<<<<<<<< 终端用户:" + account + " 心跳超过三次无应答,已被关闭 <<<<<<<<<<<<<<<<<<<<");
        }
    }
}

心跳机制实现及处理类完毕。

具体的如何配置,请查看下一章节。待续

原文链接:加载失败,请重新获取