从mina中学习超时程序编写

时间:2021-06-07 08:26:59

从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;

}

}

}

}

}