Apache Mina（二），apachemina
Apache Mina（二），apachemina
在mina的源码，整个框架最核心的几个包是 ：
- org.apache.mina.core.service ：IoService、IoProcessor、IoHandler、IoAcceptor、IoConnector
- org.apache.mina.core.session
- org.apache.mina.core.polling
- org.apache.mina.transport.socket
IoService ：Base interface for all IoAcceptors and IoConnectors that provide I/O service and manage IoSessions.
它是所有 IoAcceptor 和 IoConnector 的基接口，对于一个 IoService，有哪些信息需要我们关注呢？
IoProcessor<S extends IoSession> ：An internal interface to represent an 'I/O processor' that performs actual I/O operations for IoSessions. It abstracts existing reactor frameworks such as Java NIO once again to simplify transport implementations.
一个内部接口代表一个I/O处理器，它为 IoSession 执行实际的I/O操作。它抽象现有反应器框架，如 Java NIO 再一次简化传输实现。
1）NioSocketAcceptor 构造函数： 定义了SessionConfig，并把this传递到SessionConfig当中，指定NioProcessor。 AbstractPollingIoAcceptor 构造函数： new SimpleIoProcessorPool<S>(processorClass) ：是把NioProcessor包装成了pool.看类图IoProcessor就很好理解了，这是一个组成模式。 init（）的调用实际上是NioSocketAcceptor的，init() 源码：selector = Selector.open(); 值得注意的是，服务端在创建NioSocketAcceptor实现时，会生成一个线程池（AbstractIoService.executor），此线程池用来执行一个接受请求的任务，这个任务是AbstractPollingIoAcceptor的Acceptor，Acceptor会开一个Selector，用来监听NIO中的ACCEPT事件。任务初始化时并没有执行，而在调用NioSocketAcceptor实例的bind方法时，则会启动对指定端口的ACCEPT事件的监听。 SimpleIoProcessorPool是在NioSocketAcceptor实例化时创建的，其上有N+1（N=CPU的个数）个NIOProcessor（IoProcessor<S>[] pool）来处理实际IO的读写事件，每个pool都是从NioSocketAcceptor构造函数传过去的NioProcessor实例，并在NioProcessor构造函数传入一个线程池。 每个NIOProcessor都会对应一个Selector和 NioSocketAcceptor.init() 中的Selector 一起构成了Mina独有的双Selector模型，这种设计的优点是不会导致阻塞)，来监听Socket中的读写事件。实际对读写的操作也是在一个SimpleIoProcessorPool实例化好的一个线程池中以任务的形式执行，这个任务叫Processor(可以在AbstractPollingIoProcessor类中找到其实现) AbstractIoAcceptor 构造函数： AbstractIoAcceptor主要用来绑定监听端口。这个构造函数没有干其他的事情。 这个构造函数还有一个监听器，或者叫监听池（可以包含多个监听器）。用来监听service创建、连接、断开等动作，当上述动作发生地时候，会调用listener。里面可以写自己的一些方法。 一次请求的过程如下 ： Client通过Socket连接服务器，先是由Acceptor接收到请求连接的事件（即ACCEPT事件）。此事件由Acceptor进行处理，会创建一条Socket连接，并将此连接和一个NIOProcessor关联，这个过程通过图中的连接分配器进行，连接分配器会均衡的将Socket和不同的NIOProcessor绑定(轮流分配)，绑定完成后，会在NIOProcessor上进行读写事件的监听，而读写的实际处理则分配给Processor任务完成。当有读写事件发生时，就会通知到对应的Processor进行数据处理。 IoAcceptor acceptor = new NioSocketAcceptor();
acceptor.setHandler(new IoHandlerAdapter());
acceptor.bind(new InetSocketAddress(6969));
public NioSocketAcceptor() {
super(new DefaultSocketSessionConfig(), NioProcessor.class);
((DefaultSocketSessionConfig) getSessionConfig()).init(this);
}
protected AbstractPollingIoAcceptor(IoSessionConfig sessionConfig, Class<? extends IoProcessor<S>> processorClass) {
this(sessionConfig, null, new SimpleIoProcessorPool<S>(processorClass), true);
}
private AbstractPollingIoAcceptor(IoSessionConfig sessionConfig, Executor executor, IoProcessor<S> processor,
boolean createdProcessor) {
super(sessionConfig, executor);
if (processor == null) {
throw new IllegalArgumentException("processor");
}
this.processor = processor;
this.createdProcessor = createdProcessor;
try {
// Initialize the selector
init();
// The selector is now ready, we can switch the
// flag to true so that incoming connection can be accepted
selectable = true;
} catch (RuntimeException e) {
throw e;
} catch (Exception e) {
throw new RuntimeIoException("Failed to initialize.", e);
} finally {
if (!selectable) {
try {
destroy();
} catch (Exception e) {
ExceptionMonitor.getInstance().exceptionCaught(e);
}
}
}
}
public SimpleIoProcessorPool(Class<? extends IoProcessor<S>> processorType, Executor executor) {
this(processorType, executor, DEFAULT_SIZE); // int DEFAULT_SIZE = Runtime.getRuntime().availableProcessors() + 1;
}
public SimpleIoProcessorPool(Class<? extends IoProcessor<S>> processorType, Executor executor, int size) {
if (processorType == null) {
throw new IllegalArgumentException("processorType");
}
if (size <= 0) {
throw new IllegalArgumentException("size: " + size + " (expected: positive integer)");
}
// Create the executor if none is provided
createdExecutor = (executor == null);
if (createdExecutor) {
this.executor = Executors.newCachedThreadPool();
// Set a default reject handler
((ThreadPoolExecutor) this.executor).setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
} else {
this.executor = executor;
}
pool = new IoProcessor[size];
boolean success = false;
Constructor<? extends IoProcessor<S>> processorConstructor = null;
boolean usesExecutorArg = true;
try {
// We create at least one processor
try {
try {
processorConstructor = processorType.getConstructor(ExecutorService.class);
pool[0] = processorConstructor.newInstance(this.executor);
} catch (NoSuchMethodException e1) {
// To the next step...
try {
processorConstructor = processorType.getConstructor(Executor.class);
pool[0] = processorConstructor.newInstance(this.executor);
} catch (NoSuchMethodException e2) {
// To the next step...
try {
processorConstructor = processorType.getConstructor();
usesExecutorArg = false;
pool[0] = processorConstructor.newInstance();
} catch (NoSuchMethodException e3) {
// To the next step...
}
}
}
} catch (RuntimeException re) {
LOGGER.error("Cannot create an IoProcessor :{}", re.getMessage());
throw re;
} catch (Exception e) {
String msg = "Failed to create a new instance of " + processorType.getName() + ":" + e.getMessage();
LOGGER.error(msg, e);
throw new RuntimeIoException(msg, e);
}
if (processorConstructor == null) {
// Raise an exception if no proper constructor is found.
String msg = 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.";
LOGGER.error(msg);
throw new IllegalArgumentException(msg);
}
// Constructor found now use it for all subsequent instantiations
for (int i = 1; i < pool.length; i++) {
try {
if (usesExecutorArg) {
pool[i] = processorConstructor.newInstance(this.executor);
} else {
pool[i] = processorConstructor.newInstance();
}
} catch (Exception e) {
// Won't happen because it has been done previously
}
}
success = true;
} finally {
if (!success) {
dispose();
}
}
}
public NioProcessor(Executor executor) {
super(executor);
try {
// Open a new selector
selector = Selector.open();
} catch (IOException e) {
throw new RuntimeIoException("Failed to open a selector.", e);
}
}
protected AbstractIoAcceptor(IoSessionConfig sessionConfig, Executor executor) {
super(sessionConfig, executor);
defaultLocalAddresses.add(null);
}
protected AbstractIoService(IoSessionConfig sessionConfig, Executor executor) {
if (sessionConfig == null) {
throw new IllegalArgumentException("sessionConfig");
}
if (getTransportMetadata() == null) {
throw new IllegalArgumentException("TransportMetadata");
}
if (!getTransportMetadata().getSessionConfigType().isAssignableFrom(sessionConfig.getClass())) {
throw new IllegalArgumentException("sessionConfig type: " + sessionConfig.getClass() + " (expected: "
+ getTransportMetadata().getSessionConfigType() + ")");
}
// Create the listeners, and add a first listener : a activation listener
// for this service, which will give information on the service state.
listeners = new IoServiceListenerSupport(this);
listeners.add(serviceActivationListener);
// Stores the given session configuration
this.sessionConfig = sessionConfig;
// Make JVM load the exception monitor before some transports
// change the thread context class loader.
ExceptionMonitor.getInstance();
if (executor == null) {
this.executor = Executors.newCachedThreadPool();
createdExecutor = true;
} else {
this.executor = executor;
createdExecutor = false;
}
threadName = getClass().getSimpleName() + '-' + id.incrementAndGet();
}
评论暂时关闭