从mina中学习超时程序编写
从mina中学习超时程序编写
在很多情况下，程序需要使用计时器定，在指定的时间内检查连接过期。例如，要实现一个mqtt服务，为了保证QOS，在服务端发送消息后，需要等待客户端的ack，确保客户端接收到消息，当服务端等待一段时间后，仍未获得客户端ack，就会将消息重新传送。在Mina中，每个链接都可以设置read ideal 和write ideal 时间，当链接read ideal或者write ideal超时后，可以触发用户自定义的一个动作（比如关闭链接等），本文主要是探讨mina中如何实现这种超时策略，也给大家带来参考（IndexIdleChecker.java）。
1. 主要的数据结构
在IndexIdleChecker中，有如下属性定义：
private final Set<AbstractIoSession>[] readIdleSessionIndex = new Set[MAX_IDLE_TIME_IN_SEC];
private final Set<AbstractIoSession>[] writeIdleSessionIndex =new Set[MAX_IDLE_TIME_IN_SEC];
分别用于记录readIdle和writeIdle，以下仅以readIdle为例，因为writeIdle与readIdle流程基本一样。
注意到readIdleSessionIndex的类型是数组，并且数组中的每个元素是Set类型，Set中的每一个元素都是Session（AbstractIoSession的简写）类型，readIdle数组实际大小为3600，mina默认的session最大过期时间为一小时，因此，数组的每一个元素记录了这一秒内将要过期的session集合（注意，算法使用的是循环数组，因为默认最多一个小时就过期）。
IndexIdleChecker还有一个属性：
private long lastCheckTimeMs = System.currentTimeMillis();
用于记录上次处理过期请求的时间。
每个session对象中都有一个property，用于记录这个session目前在readIdle数组中的位置（READ_IDLE_INDEX）。
2. Read（或者write）请求ideal处理策略
Read事件到来的方法签名如下：
public void sessionRead(AbstractIoSession session, long timeInMs) ;
先通过session的READ_IDLE_INDEX 属性，获取session在readIdeal中的位置（oldIndex），之后删除readIdeal中对应oldIndex的set中的当前session，然后计算当前session下次的过期时间，将过期时间%3600后，求出其对应的readIdeal数组下标（index），将session放入readIdeal的index位置的set中，并且设置session的READ_IDLE_INDEX 的属性值为index。
3. Worker线程
在运行时Mina会启动一个Worker守护线程，代码如下：
@Override
public void run() {
while (running) {
try {
sleep(GRANULARITY_IN_MS);
processIdleSession(System.currentTimeMillis());
} catch (InterruptedException e) {
break;
}
}
}
Worker进程在每次处理完session过期后，都会sleep 1s，然后进行下一次的过期处理，在processIdleSession方法中，仅需每次处理
int startIdx = ((int) (Math.max(lastCheckTimeMs, timeMs -MAX_IDLE_TIME_IN_MS + 1) / 1000L))
% MAX_IDLE_TIME_IN_SEC;
int endIdx = ((int) (timeMs / 1000L)) %MAX_IDLE_TIME_IN_SEC;
startIdx和endIdx之间的readIdeal数组即可。针对每个idx，拿出set，将set中的所有session的READ_IDLE_INDEX 置为null，并调用session中的用户处理过程即可（比如关闭session）。（注意，如果用户设置过期时间为-1，表示永远不过期，此时不做任何处理）。需要注意的是，worker线程每次至少会处理一个index，有时会处理多个index，比如，当前这次处理index的时间超过了3s，之后worker又sleep 1s，那么下次worker被唤醒，将处理这4s内的4个index。
小结：
在大多数情况下，链接线程比较活跃的情况下，session的存储位置会在数组中不断的向后移动（因为是循环数组，所以没有边界），因此当worker要处理启动处理过期时，活跃的session的index一定会在过期处理的index之前，因此，仅有那些不活跃的到期的session才会被查询和处理掉，worker的处理代价并不高。
使用这种方式不需要计时器等支持，实现简单，是一种比较好的超时处理方式。
参考：
<dependency>
<groupId>org.apache.mina</groupId>
<artifactId>mina-core</artifactId>
<version>3.0.0-M2</version>
</dependency>
源码：
public class IndexedIdleChecker implements IdleChecker {
/**Maximum idle time in second : default to 1 hour */
private static final int MAX_IDLE_TIME_IN_SEC = 60 * 60;
/**Maximum idle time in milliseconds : default to 1 hour */
private static final long MAX_IDLE_TIME_IN_MS =MAX_IDLE_TIME_IN_SEC * 1000L;
/** Alogger for this class */
private static final Logger LOG = LoggerFactory.getLogger(IndexedIdleChecker.class);
// Aspeedup for logs
private static final boolean IS_DEBUG = LOG.isDebugEnabled();
private static final AttributeKey<Integer>READ_IDLE_INDEX = AttributeKey.createKey(Integer.class,
"idle.read.index");
private static final AttributeKey<Integer>WRITE_IDLE_INDEX = AttributeKey.createKey(Integer.class,
"idle.write.index");
private long lastCheckTimeMs = System.currentTimeMillis();
@SuppressWarnings("unchecked")
private final Set<AbstractIoSession>[] readIdleSessionIndex =new Set[MAX_IDLE_TIME_IN_SEC];
@SuppressWarnings("unchecked")
private final Set<AbstractIoSession>[] writeIdleSessionIndex =new Set[MAX_IDLE_TIME_IN_SEC];
/** Theelapsed period between two checks : 1 second */
private static final int GRANULARITY_IN_MS = 1000;
private final Worker worker = new Worker();
private volatile boolean running = true;
/**
* {@inheritDoc}
*/
@Override
public void start() {
worker.start();
}
/**
* {@inheritDoc}
*/
@Override
public void destroy() {
running = false;
try {
// interrupt the sleep
worker.interrupt();
// wait for worker to stop
worker.join();
} catch (InterruptedException e) {
// interrupted, we don't care much
}
}
/**
* {@inheritDoc}
*/
@Override
public void sessionRead(AbstractIoSession session, long timeInMs) {
if (IS_DEBUG) {
LOG.debug("session read event, compute idle index of session {}", session);
}
// remove from the old index position
Integer oldIndex =session.getAttribute(READ_IDLE_INDEX);
if (oldIndex !=null && readIdleSessionIndex[oldIndex] != null) {
if (IS_DEBUG) {
LOG.debug("remove for old index {}", oldIndex);
}
readIdleSessionIndex[oldIndex].remove(session);
}
long idleTimeInMs = session.getConfig().getIdleTimeInMillis(IdleStatus.READ_IDLE);
// is idle enabled ?
if (idleTimeInMs <= 0L) {
if (IS_DEBUG) {
LOG.debug("no read idle configuration");
}
} else {
int nextIdleTimeInSeconds = (int) ((timeInMs +idleTimeInMs) / 1000L);
int index = nextIdleTimeInSeconds %MAX_IDLE_TIME_IN_SEC;
if (IS_DEBUG) {
LOG.debug("computed index : {}", index);
}
if (readIdleSessionIndex[index] ==null) {
readIdleSessionIndex[index] =Collections
.newSetFromMap(newConcurrentHashMap<AbstractIoSession, Boolean>());
}
if (IS_DEBUG) {
LOG.debug("marking session {} idle for index {}", session, index);
}
readIdleSessionIndex[index].add(session);
session.setAttribute(READ_IDLE_INDEX, index);
}
}
/**
* {@inheritDoc}
*/
@Override
public void sessionWritten(AbstractIoSession session, long timeInMs) {
if (IS_DEBUG) {
LOG.debug("session write event, compute idle index of session {}", session);
}
// remove from the old index position
Integer oldIndex =session.getAttribute(WRITE_IDLE_INDEX);
if (oldIndex !=null && writeIdleSessionIndex[oldIndex] != null) {
if (IS_DEBUG) {
LOG.debug("remove for old index {}", oldIndex);
}
writeIdleSessionIndex[oldIndex].remove(session);
}
long idleTimeInMs = session.getConfig().getIdleTimeInMillis(IdleStatus.WRITE_IDLE);
// is idle enabled ?
if (idleTimeInMs <= 0L) {
if (IS_DEBUG) {
LOG.debug("no write idle configuration");
}
} else {
int nextIdleTimeInSeconds = (int) ((timeInMs +idleTimeInMs) / 1000L);
int index = nextIdleTimeInSeconds %MAX_IDLE_TIME_IN_SEC;
if (writeIdleSessionIndex[index] ==null) {
writeIdleSessionIndex[index] =Collections
.newSetFromMap(newConcurrentHashMap<AbstractIoSession, Boolean>());
}
writeIdleSessionIndex[index].add(session);
session.setAttribute(WRITE_IDLE_INDEX, index);
}
}
/**
* {@inheritDoc}
*/
@Override
public int processIdleSession(long timeMs) {
int counter = 0;
long delta = timeMs -lastCheckTimeMs;
if (LOG.isDebugEnabled()) {
LOG.debug("checking idle time, last = {}, now = {}, delta = {}",new Object[] { lastCheckTimeMs, timeMs,
delta });
}
if (delta < 1000) {
LOG.debug("not a second between the last checks, abort");
return 0;
}
// if (lastCheckTimeMs == 0) {
// LOG.debug("first check, we start now");
// lastCheckTimeMs = System.currentTimeMillis() - 1000;
// }
int startIdx = ((int) (Math.max(lastCheckTimeMs, timeMs -MAX_IDLE_TIME_IN_MS + 1) / 1000L))
% MAX_IDLE_TIME_IN_SEC;
int endIdx = ((int) (timeMs / 1000L)) %MAX_IDLE_TIME_IN_SEC;
LOG.debug("scaning from index {} to index {}", startIdx, endIdx);
int index = startIdx;
do {
LOG.trace("scanningindex {}", index);
// look at the read idle index
counter += processIndex(readIdleSessionIndex, index, IdleStatus.READ_IDLE);
counter += processIndex(writeIdleSessionIndex, index, IdleStatus.WRITE_IDLE);
index = (index + 1) % MAX_IDLE_TIME_IN_SEC;
} while (index != endIdx);
// save last check time for next call
lastCheckTimeMs = timeMs;
LOG.debug("detected {} idleing sessions", counter);
return counter;
}
private int processIndex(Set<AbstractIoSession>[]indexByTime, int position, IdleStatus status) {
Set<AbstractIoSession> sessions =indexByTime[position];
if (sessions ==null) {
return 0;
}
int counter = 0;
for (AbstractIoSession idleSession : sessions) {
idleSession.setAttribute(status ==IdleStatus.READ_IDLE ?READ_IDLE_INDEX : WRITE_IDLE_INDEX, null);
// check if idle detection wasn't disabled since the index update
if (idleSession.getConfig().getIdleTimeInMillis(status)> 0) {
idleSession.processSessionIdle(status);
}
counter++;
}
// clear the processed index entry
indexByTime[position] = null;
return counter;
}
/**
* Thread in charge of checking the idleingsessions and fire events
*/
private class Worker extends Thread {
public Worker() {
super("IdleChecker");
setDaemon(true);
}
@Override
public void run() {
while (running) {
try {
sleep(GRANULARITY_IN_MS);
processIdleSession(System.currentTimeMillis());
} catch (InterruptedException e) {
break;
}
}
}
}
}
