NIO（三）：源码分析_初始化部分

骑猪看日落 2023-06-27 05:42 10阅读 0赞

# 1，ServerSocketChannel初始化 #

## 1.1，ServerSocketChannel serverSocketChannel = ServerSocketChannel.open() ##

*  `ServerSocketChannel.open()`

public static ServerSocketChannel open() throws IOException {
        // 先获取一个Provider
        // 再通过Provider打开Channel
        return SelectorProvider.provider().openServerSocketChannel();
    }

*  Selector.provider()

/**********Provider 根据平台创建, (Windows/Linux)************/
    public static SelectorProvider provider() {
        synchronized (lock) {
            if (provider != null)
                return provider;
            return AccessController.doPrivileged(
                new PrivilegedAction<SelectorProvider>() {
                    public SelectorProvider run() {
                        if (loadProviderFromProperty())
                            return provider;
                        if (loadProviderAsService())
                            return provider;
                        provider = sun.nio.ch.DefaultSelectorProvider.create();
                        return provider;
                    }
                });
        }
    }
    /******创建一个Windows平台的Provider*******/
    public static SelectorProvider create() {
        return new WindowsSelectorProvider();
    }

*  `WindowsSelectorProvider`继承自`SelectorProviderImpl`，`openServerSocketChannel()`为父类方法

/****SelectorProviderImpl***/
    public ServerSocketChannel openServerSocketChannel() throws IOException {
        // 所以实际构建的实例为ServerSocketChannelImpl
        return new ServerSocketChannelImpl(this);
    }
    
    /**ServerSocketChannelImpl***/
    ServerSocketChannelImpl(SelectorProvider var1) throws IOException {
        super(var1);
        // 注意此处默认阻塞方式为true
        // 此处初始化文件描述符:FileDescriptor,用来表示公开的套接字
        this.fd = Net.serverSocket(true);
        this.fdVal = IOUtil.fdVal(this.fd);
        this.state = 0;
    }

## 1.2，`serverSocketChannel.configureBlocking(false)`：修改为非阻塞 ##

*  `AbstractSelectableChannel.configureBlocking`()

public final SelectableChannel configureBlocking(boolean block) throws IOException {
        synchronized (regLock) {
            if (!isOpen())
                throw new ClosedChannelException();
            if (blocking == block)
                return this;
            // 初始化阻塞并且存在有效的注册事件,不允许设置, 直接异常处理
            if (block && haveValidKeys())
                throw new IllegalBlockingModeException();
            // 该方法为抽象方法, 上一步初始化ServerSocketChannel实际类型为ServerSocketChannelImpl，直接去改方法处理
            implConfigureBlocking(block);
            blocking = block;
        }
        return this;
    }

*  `AbstractSelectableChannel.haveValidKeys()`：校验当前通道中是否存在注册事件

private boolean haveValidKeys() {
        synchronized (keyLock) {
            // 不存在任务,直接返回false
            if (keyCount == 0)
                return false;
            // 存在注册事件,并且注册事件未失效,返回true,
            for (int i = 0; i < keys.length; i++) {
                if ((keys[i] != null) && keys[i].isValid())
                    return true;
            }
            return false;
        }
    }

*  ServerSocketChannelImpl.implConfigureBlocking()

protected void implConfigureBlocking(boolean var1) throws IOException {
        // 通过native方法，修改文件描述符fd的阻塞方式为设置方式
        IOUtil.configureBlocking(this.fd, var1);
    }

## 1.3，`serverSocketChannel.socket().bind(new InetSocketAddress(8080))` ##

*  `serverSocketChannel.socket()`：获取`Socket`

public ServerSocket socket() {
        Object var1 = this.stateLock;
        // 该对象通过单例获取
        synchronized(this.stateLock) {
            if(this.socket == null) {
                // 使用ServerSocketAdaptor对象进行构建
                this.socket = ServerSocketAdaptor.create(this);
            }
    
            return this.socket;
        }
    }

*  `ServerSocketAdaptor.create(this)`

// 成员变量， 进行ServerSocketChannel存储
    private final ServerSocketChannelImpl ssc;
    
    public static ServerSocket create(ServerSocketChannelImpl var0) {
        try {
            // 实际构造的Socket对象是ServerSocketAdaptor
            // 同时传递Channel到Socket
            // 实现互相获取
            return new ServerSocketAdaptor(var0);
        } catch (IOException var2) {
            throw new Error(var2);
        }
    }
    
    private ServerSocketAdaptor(ServerSocketChannelImpl var1) throws IOException {
        this.ssc = var1;
    }

*  `Socket.bind(new InetSocketAddress(8080))`：绑定端口，从上一步可知，初始化的`Socket`类型为`ServerSocketAdapter`

public void bind(SocketAddress var1, int var2) throws IOException {
        if(var1 == null) {
            var1 = new InetSocketAddress(0);
        }
        try {
            // Socket初始化时，传递Channel对象并赋值给ssc属性
            // 所以ssc代表的是派生该Socket的Channel对象
            // 此处转一圈后依旧调用Channel.bind()方法，与serverSocketChannel.bind()直接绑定效果一致
            this.ssc.bind((SocketAddress)var1, var2);
        } catch (Exception var4) {
            Net.translateException(var4);
        }
    }

*  `Channel.bind()`：`Channel`初始化的实例为`ServerSocketChannelImpl`

public ServerSocketChannel bind(SocketAddress var1, int var2) throws IOException {
        Object var3 = this.lock;
        // 串行化执行,进行加锁
        synchronized(this.lock) {
            if(!this.isOpen()) { // 通道不是打开状态,异常处理
                throw new ClosedChannelException();
            } else if(this.isBound()) { // 已经绑定过,不允许多次绑定
                throw new AlreadyBoundException();
            } else {
                // 此处对初始化的地址进行判断,
                // 如果没有初始化,直接初始化0端口,表示随机端口
                // 如果已经初始化,则对IP地址进行判断
                InetSocketAddress var4 = var1 == null?new InetSocketAddress(0):Net.checkAddress(var1);
                // 通过安全管理器对端口监听进行校验,不是很懂
                SecurityManager var5 = System.getSecurityManager();
                if(var5 != null) {
                    var5.checkListen(var4.getPort());
                }
                NetHooks.beforeTcpBind(this.fd, var4.getAddress(), var4.getPort());
                // 重点关注下面两个方法
                // 把IP和端口全部绑定到fd中去
                Net.bind(this.fd, var4.getAddress(), var4.getPort());
                // 并进行监听
                Net.listen(this.fd, var2 < 1?50:var2);
                Object var6 = this.stateLock;
                synchronized(this.stateLock) {
                    // 启动完成后,初始化localAddress, 再次启动会直接报错
                    // 主要是从FD中拿到绑定的IP和端口存储到InetSocketAddress中
                    this.localAddress = Net.localAddress(this.fd);
                }
    
                return this;
            }
        }
    }

*  `Channel.isBound()`：绑定情况判断

public boolean isBound() {
        Object var1 = this.stateLock;
        synchronized(this.stateLock) {
            // localAddress为InetSocketAddress,不为null说明已经初始化
            // 此处表示IP:端口绑定不允许多次绑定
            return this.localAddress != null;
        }
    }

# 2，SocketChannel初始化 #

2.1，`SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress("127.0.0.1", 8080))`

*  `SocketChannel.open(new InetSocketAddress("127.0.0.1", 8080))`

public static SocketChannel open(SocketAddress remote) throws IOException {
        // 获取SocketChannel实例对象
        SocketChannel sc = open();
        try {
            // 连接当前客户端到服务器
            sc.connect(remote);
        } catch (Throwable x) {
            try {
                sc.close();
            } catch (Throwable suppressed) {
                x.addSuppressed(suppressed);
            }
            throw x;
        }
        assert sc.isConnected();
        return sc;
    }

*  `SocketChannel.open()`：最终获取`SocketChannelImpl`

// state状态
    private static final int ST_UNINITIALIZED = -1; // 未初始化
    private static final int ST_UNCONNECTED = 0; // 未连接
    private static final int ST_PENDING = 1; // 连接中
    private static final int ST_CONNECTED = 2; // 已连接
    private static final int ST_KILLPENDING = 3; // 正在关闭
    private static final int ST_KILLED = 4; // 关闭
    
    public static SocketChannel open() throws IOException {
        // 获取WINDOWS的Provider，并打开SocketChannel
        return SelectorProvider.provider().openSocketChannel();
    }
    
    public SocketChannel openSocketChannel() throws IOException {
        // 最终构建SocketChannelImpl对象
        return new SocketChannelImpl(this);
    }
    
    SocketChannelImpl(SelectorProvider var1) throws IOException {
        // 传递Provider到父类
        super(var1);
        // 初始化文件描述符为阻塞
        this.fd = Net.socket(true);
        this.fdVal = IOUtil.fdVal(this.fd);
        // 此处注意状态变更
        this.state = 0;
    }

*  `Channel.connect()`：`Channel`为刚才初始化的`SocketChannelImpl`

// 该方法重点是进行连接， 连接完成后对state状态进行修改
    public boolean connect(SocketAddress var1) throws IOException {
        boolean var2 = false;
        Object var3 = this.readLock;
        synchronized(this.readLock) {
            Object var4 = this.writeLock;
            synchronized(this.writeLock) {
                // 对连接状态进行判断,如果已经关闭,正在连接中或者已连接,进行异常处理
                this.ensureOpenAndUnconnected();
                // 对连接的IP地址进行判断
                InetSocketAddress var5 = Net.checkAddress(var1);
                // 安全管理器校验连接,与ServerSocketChannel基本呼应吧感觉
                SecurityManager var6 = System.getSecurityManager();
                if(var6 != null) {
                    var6.checkConnect(var5.getAddress().getHostAddress(), var5.getPort());
                }
                boolean var10000;
                synchronized(this.blockingLock()) {
                    int var8 = 0;
                    Object var9;
                    try {
                        try {
                            this.begin();
                            var9 = this.stateLock;
                            synchronized(this.stateLock) {
                                if(!this.isOpen()) {
                                    boolean var10 = false;
                                    return var10;
                                }
                                // 通过localAdress保存连接IP和地址, 方便后续获取
                                if(this.localAddress == null) {
                                    NetHooks.beforeTcpConnect(this.fd, var5.getAddress(), var5.getPort());
                                }
                                this.readerThread = NativeThread.current();
                            }
                            do {
                                InetAddress var31 = var5.getAddress();
                                if(var31.isAnyLocalAddress()) {
                                    var31 = InetAddress.getLocalHost();
                                }
    							// 进行连接
                                var8 = Net.connect(this.fd, var31, var5.getPort());
                            } while(var8 == -3 && this.isOpen()); // 连接失败,循环重试
                        } finally {
                            this.readerCleanup();
                            this.end(var8 > 0 || var8 == -2);
    
                            assert IOStatus.check(var8);
    
                        }
                    } catch (IOException var27) {
                        this.close();
                        throw var27;
                    }
    				// 修改state状态
                    var9 = this.stateLock;
                    synchronized(this.stateLock) {
                        this.remoteAddress = var5;
                        // 连接结果小于0， 说明未成功， 对未成功继续进行半段
                        if(var8 <= 0) {
                            // 非阻塞模式下，可能还在连接中
                            if(!this.isBlocking()) {
                                this.state = 1;
                            // 阻塞模式下，直接失败
                            } else {
                                assert false;
                            }
                        // 连接结果大于0，连接成功，修改state状态为对应状态
                        } else {
                            this.state = 2;
                            if(this.isOpen()) {
                                this.localAddress = Net.localAddress(this.fd);
                            }
    
                            var10000 = true;
                            return var10000;
                        }
                    }
                }
    
                var10000 = false;
                return var10000;
            }
        }
    }

2.2，`socketChannel.configureBlocking(false)`：***与`ServerSocketChannel`一致***

# 3，Pipe初始化 #

*  `Pipe.open()`：初始化管道

public static Pipe open() throws IOException {
        // 继续通过Provider打开一个管道
        return SelectorProvider.provider().openPipe();
    }
    
    // 内部通过子类进行初始化
    public Pipe openPipe() throws IOException {
        return new PipeImpl(this);
    }
    
    // 此处初始化一个内部类,并交由native执行
    // 根据猜测,native方法执行后,会调用Initializer.run方法
    PipeImpl(SelectorProvider var1) throws IOException {
        try {
            AccessController.doPrivileged(new PipeImpl.Initializer(var1));
        } catch (PrivilegedActionException var3) {
            throw (IOException)var3.getCause();
        }
    }

*  Initializer.run()

public Void run() throws IOException {
        // 该方法会继续调用其内部类LoopbackConnector的run()方法
        PipeImpl.Initializer.LoopbackConnector var1 = new PipeImpl.Initializer.LoopbackConnector();
        // 该方法内部会初始化的Pipe管道的读写对象
        var1.run();
        // 初始化完成后,对异常进行分类处理
        if(this.ioe instanceof ClosedByInterruptException) {
            ...
        }
    
        if(this.ioe != null) {
            throw new IOException("Unable to establish loopback connection", this.ioe);
        } else {
            return null;
        }
    }

*  `LoopbackConnector.run()`：初始化

public void run() {
        // 定义服务端Channel
        ServerSocketChannel var1 = null;
        // 定义两个客户端Channel, 分别进行读写
        SocketChannel var2 = null;
        SocketChannel var3 = null;
    
        try {
            // 初始化两个ByteBuffer缓冲区,支持读写
            ByteBuffer var4 = ByteBuffer.allocate(16);
            ByteBuffer var5 = ByteBuffer.allocate(16);
            InetAddress var6 = InetAddress.getByName("127.0.0.1");
    
            assert var6.isLoopbackAddress();
    		// 对服务端创建的IP和端口进行存储
            InetSocketAddress var7 = null;
    		// 自旋处理, 直接成功初始化出读写对象
            while(true) 
                // 初始化ServerSocketChannel,用户提供服务
                if(var1 == null || !var1.isOpen()) {
                    var1 = ServerSocketChannel.open();
                    var1.socket().bind(new InetSocketAddress(var6, 0));
                    var7 = new InetSocketAddress(var6, var1.socket().getLocalPort());
                }
    			// 通过已经初始化的IP和端口打开一个写通道
                var2 = SocketChannel.open(var7);
            	// 此处表示随便写点东西
                PipeImpl.RANDOM_NUMBER_GENERATOR.nextBytes(var4.array());
    
                do {
                    // 将数据全部写出去
                    var2.write(var4);
                } while(var4.hasRemaining());
    
                var4.rewind();
            	// 通过服务端获取一个读请求
            	// 此处读取上一步写的数据
                var3 = var1.accept();
    
                do {
                    // 读取到写数据, 添加到另一个缓冲区
                    var3.read(var5);
                } while(var5.hasRemaining());
    
                var5.rewind();
            	// 如果读写数据一致,说明管道通信正常,初始化管道的读对象和写对象
                if(var5.equals(var4)) {
                    // 读对象
                    PipeImpl.this.source = new SourceChannelImpl(Initializer.this.sp, var2);
                    // 写对象
                    PipeImpl.this.sink = new SinkChannelImpl(Initializer.this.sp, var3);
                    break;
                }
    
                var3.close();
                var2.close();
            }
        } catch (IOException var18) {
            ...
    		// IO异常后,初始化IOE
            Initializer.this.ioe = var18;
        } finally { ... }
    }

# 4，Selector初始化 #

## 4.1，`selector =` `Selector.open()` ##

public static Selector open() throws IOException {
        // 初始化WINDOWS系统的Provider, 初始化轮询器
        return SelectorProvider.provider().openSelector();
    }

## 4.2，`WindowsSelectorProvider.openSelector()` ##

*  WindowsSelectorProvider.openSelector()

public AbstractSelector openSelector() throws IOException {
        // 注意此处的轮询器, 是平台相关, WINDOWS下, 初始化为WindowsSelectorImpl
        return new WindowsSelectorImpl(this);
    }
    
    /***** WindowsSelectorImpl初始化部分 *****/
    // 初始化管道
    private final Pipe wakeupPipe = Pipe.open();
    // 初始化pollWrapper
    private PollArrayWrapper pollWrapper = new PollArrayWrapper(8);
    WindowsSelectorImpl(SelectorProvider var1) throws IOException {
        // 先初始化父类,即SelectorImpl
        super(var1);
        // 保存SourceChannel的Socket句柄
        this.wakeupSourceFd = ((SelChImpl)this.wakeupPipe.source()).getFDVal();
        // 保存SinkChannel的Socket句柄
        SinkChannelImpl var2 = (SinkChannelImpl)this.wakeupPipe.sink();
        var2.sc.socket().setTcpNoDelay(true);
        this.wakeupSinkFd = var2.getFDVal();
        // pollWrapper: 用于存储Socket句柄FD和事件Events
        // addWakeupSocket: 表示添加一个Socket事件
        // 此处意思就是将Socket事件添加到内存空间中
        this.pollWrapper.addWakeupSocket(this.wakeupSourceFd, 0);
    }
    
    // 父类初始化了五个keys集合
    // 已选择集合,select()时添加到该集合并返回
    protected Set<SelectionKey> selectedKeys = new HashSet();
    // register()时添加到该集合, 表示所有注册过的事件
    protected HashSet<SelectionKey> keys = new HashSet();
    // 将该集合与keys关联
    private Set<SelectionKey> publicKeys;
    // 将该集合与selectedKeys关联
    private Set<SelectionKey> publicSelectedKeys;
    // 已经取消事件
    private final Set<SelectionKey> cancelledKeys = new HashSet<SelectionKey>();
    
    protected SelectorImpl(SelectorProvider var1) {
        // 传递到父类AbstractSelector持有Provider
        super(var1);
        // 关联key
        if(Util.atBugLevel("1.4")) {
            this.publicKeys = this.keys;
            this.publicSelectedKeys = this.selectedKeys;
        } else {
            this.publicKeys = Collections.unmodifiableSet(this.keys);
            this.publicSelectedKeys = Util.ungrowableSet(this.selectedKeys);
        }
    }

*  `PollArrayWrapper`初始化

// 内存空间操作对象
    private AllocatedNativeObject pollArray;
    // 表示内存空间地址
    long pollArrayAddress;
    
    PollArrayWrapper(int var1) {
        int var2 = var1 * SIZE_POLLFD;
        // 初始化时通过unsafe申请到一块内存空间
        this.pollArray = new AllocatedNativeObject(var2, true);
        this.pollArrayAddress = this.pollArray.address();
        // 初始化Socket句柄大小
        this.size = var1;
    }

*  `PollArrayWrapper.addWakeupSocket()`：添加一个事件

void addWakeupSocket(int var1, int var2) {
        // 表示Socket句柄位置填充
        this.putDescriptor(var2, var1);
        // 表示Socket事件位置填充
        this.putEventOps(var2, Net.POLLIN);
    }
    
    void putDescriptor(int var1, int var2) {
        // Socket句柄占用四个字节
        this.pollArray.putInt(SIZE_POLLFD * var1 + 0, var2);
    }
    
    void putEventOps(int var1, int var2) {
        // Socket事件占用两个字节
        this.pollArray.putShort(SIZE_POLLFD * var1 + 4, (short)var2);
    }

# 5，ByteBuffer初始化 #

*  `ByteBuffer`之前已经有过分析，主要是通过四个基本属性进行缓冲区数据读写和标记，此处主要关注直接缓冲区和间接缓冲区的初始化方式
 *  间接缓冲区

public static ByteBuffer allocate(int capacity) {
        if (capacity < 0)
            throw new IllegalArgumentException();
        // 间接缓冲区, 初始化为类型为HeapByteBuffer
        return new HeapByteBuffer(capacity, capacity);
    }
    
    HeapByteBuffer(int cap, int lim) {
    	// 参数顺序依次为: mark，position，limit，capacity，底层数据，offset偏移量
        super(-1, 0, lim, cap, new byte[cap], 0);
    }

*  直接缓冲区

public static ByteBuffer allocateDirect(int capacity) {
        // 直接缓冲区初始化实例为 DirectByteBuffer
        // 直接缓冲区直接从内存中申请空间, 直接进行数据操作
        return new DirectByteBuffer(capacity);
    }

*  其他`ByteBuffer`读写无非是对四个属性位置进行移动，并从底层数组中获取数据，注意NIO的`ByteBuffer`不会扩容