Mina之service

1、TransportMetadata定义了一组操作底层的元数据信息,它的一个实现是DefaultTransportMetadata,维护的信息如下:

    private final String providerName;
private final String name;
private final boolean connectionless;
private final boolean fragmentation;
private final Class<? extends SocketAddress> addressType;
private final Class<? extends IoSessionConfig> sessionConfigType;
private final Set<Class<? extends Object>> envelopeTypes;

2、IoSessionConfig定义一组对会话配置的操作,包括了读写缓冲区的大小的设置、读写间隔的调整、失效时限等。抽象类AbstractIoSessionConfig中定义了最基本的信息(各种I/O的共性)如下:

    private int minReadBufferSize = 64;
private int readBufferSize = 2048;
private int maxReadBufferSize = 65536;
private int idleTimeForRead;
private int idleTimeForWrite;
private int idleTimeForBoth;
private int writeTimeout = 60;
private boolean useReadOperation;
private int throughputCalculationInterval = 3;

因为可用于不同的I/O,我们这里只关心下Socket,也就是AbstractSocketSessionConfig(增加与Socket相关操作):

public abstract class AbstractSocketSessionConfig extends AbstractIoSessionConfig
implements SocketSessionConfig

再下面的特殊的类比如NioSocketSession就可以实现一个内部类,继承AbstractSocketSessionConfig后得到一组接口操作这些配置项。

3、IoServiceListener与这个服务相关事件的监听者,定义如下:

public interface IoServiceListener extends EventListener {
void serviceActivated(IoService service) throws Exception;
void serviceIdle(IoService service, IdleStatus idleStatus) throws Exception;
void serviceDeactivated(IoService service) throws Exception;
void sessionCreated(IoSession session) throws Exception;
void sessionDestroyed(IoSession session) throws Exception;
}

该接口用作在后面生成匿名类的时候用,而IoServiceListenerSupport负责将IoService和IoServiceListener包装到一起进行管理:

    private final IoService service;
private final List<IoServiceListener> listeners = new CopyOnWriteArrayList<IoServiceListener>();
private final ConcurrentMap<Long, IoSession> managedSessions = new ConcurrentHashMap<Long, IoSession>();
private final Map<Long, IoSession> readOnlyManagedSessions = Collections.unmodifiableMap(managedSessions);
private final AtomicBoolean activated = new AtomicBoolean();
private volatile long activationTime;
private volatile int largestManagedSessionCount;
private volatile long cumulativeManagedSessionCount;

下面是一个例子,在创建会话时通过下面的函数调用IoServiceListener的sessionCreated方法:

    public void fireSessionCreated(IoSession session) {
boolean firstSession = false;
if (session.getService() instanceof IoConnector) {
synchronized (managedSessions) {
firstSession = managedSessions.isEmpty();
}
}
if (managedSessions.putIfAbsent(Long.valueOf(session.getId()), session) != null) {
return;
}
if (firstSession) {
fireServiceActivated();
}
IoFilterChain filterChain = session.getFilterChain();
filterChain.fireSessionCreated();
filterChain.fireSessionOpened();

int managedSessionCount = managedSessions.size();
if (managedSessionCount > largestManagedSessionCount) {
largestManagedSessionCount = managedSessionCount;
}
cumulativeManagedSessionCount ++;
for (IoServiceListener l : listeners) {
try {
l.sessionCreated(session);
} catch (Throwable e) {
ExceptionMonitor.getInstance().exceptionCaught(e);
}
}
}

4、IoHandler是处理这个服务所管理的所有连接的处理器,定义了对不同动作的各个操作:

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

要注意的是I/O Processor线程是负责处理多个会话的。IoHandlerAdapter使用适配器模式对IoHandler简单封装。

5、IoFilterChain是一个监视器链的类型,定义了在其上的一组操作,监听器被放在链表上以<name-filter>的形式组织起来。它是通过IoFilterChainBuilder来建立的,而IoFilterChainBuilder的实现是DefaultIoFilterChainBuilder,其中只有一个属性:

private final List<Entry> entries;

DefaultIoFilterChainBuilder中的实现也比较简单,就不细说啦。

6、服务管理的是多个会话,所以IoService中保存的就是会话相关的操作以及服务的启动等。而IoAcceptor用于绑定到指定的IP和端口,从而接受来自客户端的连接请求,同时fire相应的客户端连接的各种事件给IoHandle去处理。而IoConnect是客户端的连接发起者接口,用于尝试连接到服务器指定的IP和端口。

7、IoAcceptor和IoConnector都是用来处理连接的,但是连接上之后真正的操作却在IoProcessor中:

public interface IoProcessor<T extends IoSession> {
boolean isDisposing();
boolean isDisposed();
void dispose();
void add(T session);
void flush(T session);
void updateTrafficControl(T session);
void remove(T session);
}

8、AbstractIoService是IoService的一个抽象基类,所包含的属性如下:

    private static final AtomicInteger id = new AtomicInteger();
private final String threadName;
private final Executor executor;//执行者
private final boolean createdExecutor;//执行者从哪来的?
private IoHandler handler;//处理会话
private final IoSessionConfig sessionConfig;//会话配置
private final IoServiceListener serviceActivationListener = new IoServiceListener() {//监听者
public void serviceActivated(IoService service) {
// Update lastIoTime.
AbstractIoService s = (AbstractIoService) service;
IoServiceStatistics _stats = s.getStatistics();
_stats.setLastReadTime(s.getActivationTime());
_stats.setLastWriteTime(s.getActivationTime());
_stats.setLastThroughputCalculationTime(s.getActivationTime());

}
public void serviceDeactivated(IoService service) {}
public void serviceIdle(IoService service, IdleStatus idleStatus) {}
public void sessionCreated(IoSession session) {}
public void sessionDestroyed(IoSession session) {}
};
private IoFilterChainBuilder filterChainBuilder = new DefaultIoFilterChainBuilder();//建立监听者链
private IoSessionDataStructureFactory sessionDataStructureFactory = new DefaultIoSessionDataStructureFactory();
private final IoServiceListenerSupport listeners;
protected final Object disposalLock = new Object();
private volatile boolean disposing;
private volatile boolean disposed;
private IoFuture disposalFuture;
private IoServiceStatistics stats = new IoServiceStatistics(this);

从下面代码中看到执行者从哪里来的:

if (executor == null) {    
this.executor = Executors.newCachedThreadPool();
createdExecutor = true;
} else {
this.executor = executor;
createdExecutor = false;
}

下面是释放资源的过程:

    public final void dispose() {
if (disposed) {
return;
}
IoFuture disposalFuture;
synchronized (disposalLock) {//获得释放锁
disposalFuture = this.disposalFuture;
if (!disposing) {
disposing = true;
try {
this.disposalFuture = disposalFuture = dispose0();//具体释放动作
} catch (Exception e) {
ExceptionMonitor.getInstance().exceptionCaught(e);
} finally {
if (disposalFuture == null) {
disposed = true;
}
}
}
}
if (disposalFuture != null) {//无中断等待释放动作完成
disposalFuture.awaitUninterruptibly();
}
if (createdExecutor) {//通过线程池去关闭线程
ExecutorService e = (ExecutorService) executor;
e.shutdown();
while (!e.isTerminated()) {
try {
e.awaitTermination(Integer.MAX_VALUE, TimeUnit.SECONDS);
} catch (InterruptedException e1) {
}
}
}
disposed = true;
}

初始化完成后的动作每个session都保持自己的属性映射图,在会话结束时应该设置这个AttributeMap:

((AbstractIoSession) session).setAttributeMap(session.getService()
.getSessionDataStructureFactory().getAttributeMap(session));

还应该配置写请求队列:

((AbstractIoSession) session).setWriteRequestQueue(session
.getService().getSessionDataStructureFactory().getWriteRequestQueue(session));

在会话的属性中加入SESSION_CREATED_FUTURE,这个属性会在连接真正建立后从会话中去除:

session.setAttribute(DefaultIoFilterChain.SESSION_CREATED_FUTURE,future);

9、AbstractIoAcceptor实现了IoAccepter接口,主要的成员是本地绑定地址:

private final List<SocketAddress> defaultLocalAddresses = new ArrayList<SocketAddress>();
private final List<SocketAddress> unmodifiableDefaultLocalAddresses = Collections.unmodifiableList(defaultLocalAddresses);
private final Set<SocketAddress> boundAddresses = new HashSet<SocketAddress>();
private boolean disconnectOnUnbind = true;
protected final Object bindLock = new Object();

AbstractIoConnector在AbstractIoService的基础上增加功能:在会话初始化结束时加入监听者,当监听链接请求被取消时立即结束此对话,成员如下:

private long connectTimeoutCheckInterval = 50L;//检查间隔时间
private long connectTimeoutInMillis = 60 * 1000L;
private SocketAddress defaultRemoteAddress;

下面是加入监听器的代码:

protected final void finishSessionInitialization0(
final IoSession session, IoFuture future) {
future.addListener(new IoFutureListener<ConnectFuture>() {
public void operationComplete(ConnectFuture future) {
if (future.isCanceled()) {
session.close(true);
}
}
});
}

SimpleIoProcessorPool是IoProcessor接口的基本实现类,实现了池话,将多个IoSession分布到多个IoProcessor上去管理,成员如下:

private static final int DEFAULT_SIZE = Runtime.getRuntime().availableProcessors() + 1;
private static final AttributeKey PROCESSOR = new AttributeKey(SimpleIoProcessorPool.class, "processor");
private final static Logger LOGGER = LoggerFactory.getLogger(SimpleIoProcessorPool.class);
private final IoProcessor<T>[] pool;
private final AtomicInteger processorDistributor = new AtomicInteger();
private final Executor executor;
private final boolean createdExecutor;
private final Object disposalLock = new Object();
private volatile boolean disposing;
private volatile boolean disposed;

出事话的过程如下:

pool = new IoProcessor[size];

boolean success = false;

Constructor<? extends IoProcessor<T>> processorConstructor = null;

boolean usesExecutorArg = true;

try {
try {
//有三种构造函数来构造一个Processor
try {
processorConstructor = processorType.getConstructor(ExecutorService.class);
pool[0] = processorConstructor.newInstance(executor);
} catch (NoSuchMethodException e) {}
try {
processorConstructor = processorType.getConstructor(Executor.class);
pool[0] = processorConstructor.newInstance(executor);
} catch (NoSuchMethodException e) {}
try {
processorConstructor = processorType.getConstructor();
usesExecutorArg = false;
pool[0] = processorConstructor.newInstance();
} catch (NoSuchMethodException e) {}
} catch (RuntimeException e) {
throw e;
} catch (Exception e) {
throw new RuntimeIoException("Failed to create a new instance of "+ processorType.getName(), e);
}
if (processorConstructor == null) {
throw new IllegalArgumentException(String.valueOf(processorType)
+ " must have a public constructor "
+ "with one "
+ ExecutorService.class.getSimpleName()
+ " parameter, "
+ "a public constructor with one "
+ Executor.class.getSimpleName()
+ " parameter or a public default constructor.");
}
for (int i = 1; i < pool.length; i++) {
try {
if (usesExecutorArg) {
pool[i] = processorConstructor.newInstance(executor);
} else {
pool[i] = processorConstructor.newInstance();
}
} catch (Exception e) {}
}
success = true;
} finally {
if (!success) {
dispose();
}
}
原文地址:https://www.cnblogs.com/ggzwtj/p/2210165.html