Apache Mina 源码再读2 IoSession创建过程源代码剖析

2024-02-18 06:58

本文主要是介绍Apache Mina 源码再读2 IoSession创建过程源代码剖析,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!



在调用bind()函数后,AcceptorOperationFuture 被注册到AbstractPollingIoAcceptor 类中的registerQueue 队列。在AbstractPollingIoAcceptor中存在IoProcessor 类。


/**
 * An internal interface to represent an 'I/O processor' that performs
 * actual I/O operations for {@link IoSession}s.  It abstracts existing
 * reactor frameworks such as Java NIO once again to simplify transport
 * implementations.
 *
 * @author <a href="http://mina.apache.org">Apache MINA Project</a>
 * 
 * @param <S> the type of the {@link IoSession} this processor can handle
 */
public interface IoProcessor<S extends IoSession> {}


IoProcessor 是执行IoSession 的I/O操作的内部接口类。  IoProcessor再次 抽象了在Java Nio 内部的reactor  frameworks 框架,并简化传输层的实现。


IoProcessor 实现类为SimpleIoProcessorPool 类。SImpleIoProcessorPool主要把IoSession分配到一个或者多个IoProcessor中。


An IoProcessor pool that distributes IoSessions into one or more IoProcessors. Most current transport implementations use this pool internally to perform better in a multi-core environment, and therefore, you won't need to use this pool directly unless you are running multiple IoServices in the same JVM.

If you are running multiple IoServices, you could want to share the pool among all services. To do so, you can create a new SimpleIoProcessorPool instance by yourself and provide the pool as a constructor parameter when you create the services.

This pool uses Java reflection API to create multiple IoProcessor instances. It tries to instantiate the processor in the following order:

  1. A public constructor with one ExecutorService parameter.
  2. A public constructor with one Executor parameter.
  3. A public default constructor
The following is an example for the NIO socket transport:
// Create a shared pool.SimpleIoProcessorPool<NioSession> pool = new SimpleIoProcessorPool<NioSession>(NioProcessor.class, 16);

 // Create two services that share the same pool.SocketAcceptor acceptor = new NioSocketAcceptor(pool);SocketConnector connector = new NioSocketConnector(pool)



在Acceptor 线程中,processHandles 会创建一个新的IoSession.


  /*** This method will process new sessions for the Worker class.  All* keys that have had their status updates as per the Selector.selectedKeys()* method will be processed here.  Only keys that are ready to accept* connections are handled here.* <p/>* Session objects are created by making new instances of SocketSessionImpl* and passing the session object to the SocketIoProcessor class.*/@SuppressWarnings("unchecked")private void processHandles(Iterator<H> handles) throws Exception {while (handles.hasNext()) {H handle = handles.next();handles.remove();// Associates a new created connection to a processor,// and get back a sessionS session = accept(processor, handle);//当有一个新socket链接时,返回一个IoSessionif (session == null) {continue;}initSession(session, null, null);// add the session to the SocketIoProcessor//当Acceptor发现新的socket,把socket和ioproces相关联。所有io事件由ioprocess处理session.getProcessor().add(session);}}}

  protected NioSession accept(IoProcessor<NioSession> processor, ServerSocketChannel handle) throws Exception {SelectionKey key = null;if (handle != null) {key = handle.keyFor(selector);}if ((key == null) || (!key.isValid()) || (!key.isAcceptable())) {return null;}// accept the connection from the clientSocketChannel ch = handle.accept();if (ch == null) {return null;}return new NioSocketSession(this, processor, ch);}


当accept事件触发时,会创建一个新NioSocketSession .

public abstract class NioSession extends AbstractIoSession {
    /** The NioSession processor */
    protected final IoProcessor<NioSession> processor;


    /** The communication channel */
    protected final Channel channel;


    /** The SelectionKey used for this session */
    protected SelectionKey key;


    /** The FilterChain created for this session */
    private final IoFilterChain filterChain;


    /**
     * 
     * Creates a new instance of NioSession, with its associated IoProcessor.
     * <br>
     * This method is only called by the inherited class.
     *
     * @param processor The associated IoProcessor
     */
    protected NioSession(IoProcessor<NioSession> processor, IoService service, Channel channel) {
        super(service);
        this.channel = channel;
        this.processor = processor;
        filterChain = new DefaultIoFilterChain(this);
    }

}

在抽象类NioSession中,每创建一个NioSession就会创建一个DefaultFilterChain对象。也就意味着每一个NioSession与自己独立的DefaultFilterChain相关联。


NioSession创建后,会通过initSession 方法来初始化NioSession. 


在初始化NioSession 之前,先看看一个IoSessionDataStructureFactory 工厂类。Provides data structures to a newly created session. 这个工厂类提供了创建NioSession的数据结构。NioSession中的数据结构主要包括存储自定义对象的IoSessionAttributeMap 类以及IoSession写入到缓冲区的写出队列WriteRequestQueue。


IoSessionDataStructFactory 接口源代码如下:

/*** Provides data structures to a newly created session.* * @author <a href="http://mina.apache.org">Apache MINA Project</a>*/
public interface IoSessionDataStructureFactory {/*** Returns an {@link IoSessionAttributeMap} which is going to be associated* with the specified <tt>session</tt>.  Please note that the returned* implementation must be thread-safe.*/IoSessionAttributeMap getAttributeMap(IoSession session) throws Exception;/*** Returns an {@link WriteRequest} which is going to be associated with* the specified <tt>session</tt>.  Please note that the returned* implementation must be thread-safe and robust enough to deal* with various messages types (even what you didn't expect at all),* especially when you are going to implement a priority queue which* involves {@link Comparator}.*/WriteRequestQueue getWriteRequestQueue(IoSession session) throws Exception;
}


在DefaultIoSessionDataStructureFactory 默认实现工厂类中实现了IoSessionAttributeMap 和WriteRequestQueue两个数据结构的实现。



IoSessionAttributeMap内部为一个并发安全的HashMap可以存储键值对。

 private static class DefaultIoSessionAttributeMap implements IoSessionAttributeMap {
        private final ConcurrentHashMap<Object, Object> attributes = new ConcurrentHashMap<Object, Object>(4);

}



WriteRequestQueue为一个并发安全的队列实现。

    private static class DefaultWriteRequestQueue implements WriteRequestQueue {
        /** A queue to store incoming write requests */
        private final Queue<WriteRequest> q = new ConcurrentLinkedQueue<WriteRequest>();

}


回头看一下,在initSession()初始化方法中,主要创建IoSessionAttributeMap和WriteRequestQueue队列两个结构。



初始化NioSession完成后,就需要把NioSession分配到指定的Processor线程中。


在分配SimpleIoProcessorPool中提供分配NioSession 与Processor线程相关联的策略。


    private IoProcessor<S> getProcessor(S session) {   //获取在IoSession中PROCESSOR 为键的对象IoProcessor<S> processor = (IoProcessor<S>) session.getAttribute(PROCESSOR);//如果该IoSession尚未与IoProcessor相关联,则把IoProccessor按照如下原则相关联if (processor == null) {if (disposed || disposing) {throw new IllegalStateException("A disposed processor cannot be accessed.");}//把processor和session相关联。根据session id和proceor pool 的大小来决定。 把IoSession的ID与线程池中顺序相关联processor = pool[Math.abs((int) session.getId()) % pool.length];if (processor == null) {throw new IllegalStateException("A disposed processor cannot be accessed.");}//设置PROCESSOR 的对象为IoProccessorsession.setAttributeIfAbsent(PROCESSOR, processor);}return processor;}

然后把NioSession 方法指定Processor线程的newSessions队列中,并启动Processor线程。


同时在内部存在IoServiceStatistics 类来统计IoSession相关事件。



此时,IoSession创建过程就完成了。


小结一下,NioSession创建过程。


1、Acceptor 线程,在select()中,轮询accept事件。

2、当accept事件触发时,创建NioSession对象,每一个NioSession都有一个DefaultFilterChain相关联。

3、当创建NioSession后,需要通过initSession()方法来初始化化NioSession.这里主要涉及IoSessionDataStructFactory工厂类创建IoSessionAttrubteMap 来存储用户自定义数据,创建WriteRequestQueue队列。 

4、当NioSession构建完成后,需要在SimpleIoProcessorPool来把NioSession 分配到指定的Processor线程中。

5、NioSession被添加到指定Processor线程中newSession队列中,然后启动start  Processor线程。






这篇关于Apache Mina 源码再读2 IoSession创建过程源代码剖析的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/720391

相关文章

Python中使用uv创建环境及原理举例详解

《Python中使用uv创建环境及原理举例详解》uv是Astral团队开发的高性能Python工具,整合包管理、虚拟环境、Python版本控制等功能,:本文主要介绍Python中使用uv创建环境及... 目录一、uv工具简介核心特点:二、安装uv1. 通过pip安装2. 通过脚本安装验证安装:配置镜像源(可

canal实现mysql数据同步的详细过程

《canal实现mysql数据同步的详细过程》:本文主要介绍canal实现mysql数据同步的详细过程,本文通过实例图文相结合给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的... 目录1、canal下载2、mysql同步用户创建和授权3、canal admin安装和启动4、canal

MySQL存储过程之循环遍历查询的结果集详解

《MySQL存储过程之循环遍历查询的结果集详解》:本文主要介绍MySQL存储过程之循环遍历查询的结果集,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录前言1. 表结构2. 存储过程3. 关于存储过程的SQL补充总结前言近来碰到这样一个问题:在生产上导入的数据发现

SpringBoot集成LiteFlow实现轻量级工作流引擎的详细过程

《SpringBoot集成LiteFlow实现轻量级工作流引擎的详细过程》LiteFlow是一款专注于逻辑驱动流程编排的轻量级框架,它以组件化方式快速构建和执行业务流程,有效解耦复杂业务逻辑,下面给大... 目录一、基础概念1.1 组件(Component)1.2 规则(Rule)1.3 上下文(Conte

Java中实现线程的创建和启动的方法

《Java中实现线程的创建和启动的方法》在Java中,实现线程的创建和启动是两个不同但紧密相关的概念,理解为什么要启动线程(调用start()方法)而非直接调用run()方法,是掌握多线程编程的关键,... 目录1. 线程的生命周期2. start() vs run() 的本质区别3. 为什么必须通过 st

SpringBoot整合Apache Flink的详细指南

《SpringBoot整合ApacheFlink的详细指南》这篇文章主要为大家详细介绍了SpringBoot整合ApacheFlink的详细过程,涵盖环境准备,依赖配置,代码实现及运行步骤,感兴趣的... 目录1. 背景与目标2. 环境准备2.1 开发工具2.2 技术版本3. 创建 Spring Boot

Linux中修改Apache HTTP Server(httpd)默认端口的完整指南

《Linux中修改ApacheHTTPServer(httpd)默认端口的完整指南》ApacheHTTPServer(简称httpd)是Linux系统中最常用的Web服务器之一,本文将详细介绍如何... 目录一、修改 httpd 默认端口的步骤1. 查找 httpd 配置文件路径2. 编辑配置文件3. 保存

Spring Boot 整合 Apache Flink 的详细过程

《SpringBoot整合ApacheFlink的详细过程》ApacheFlink是一个高性能的分布式流处理框架,而SpringBoot提供了快速构建企业级应用的能力,下面给大家介绍Spri... 目录Spring Boot 整合 Apache Flink 教程一、背景与目标二、环境准备三、创建项目 & 添

Macos创建python虚拟环境的详细步骤教学

《Macos创建python虚拟环境的详细步骤教学》在macOS上创建Python虚拟环境主要通过Python内置的venv模块实现,也可使用第三方工具如virtualenv,下面小编来和大家简单聊聊... 目录一、使用 python 内置 venv 模块(推荐)二、使用 virtualenv(兼容旧版 P

pytest+allure环境搭建+自动化实践过程

《pytest+allure环境搭建+自动化实践过程》:本文主要介绍pytest+allure环境搭建+自动化实践过程,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐... 目录一、pytest下载安装1.1、安装pytest1.2、检测是否安装成功二、allure下载安装2.