AOE家园

前言

上篇[【从入门到放弃-Java】并发编程-NIO使用]()简单介绍了nio的基础使用，本篇将深入源码分析nio中channel的实现。

简介

channel即通道，可以用来读、写数据，它是全双工的可以同时用来读写操作。这也是它与stream流的最大区别。

channel需要与buffer配合使用，channel通道的一端是buffer，一端是数据源实体，如文件、socket等。在nio中，通过channel的不同实现来处理 不同实体与数据buffer中的数据传输。

channel接口：

package ;import on;import ;/** * A nexus for I/O operations. * * <p> A channel represents an open connection to an entity such as a hardware * device, a file, a network socket, or a program component that is capable of * performing one or more distinct I/O operations, for example reading or * writing. * * <p> A channel is either open or closed. A channel is open upon creation, * and once closed it remains closed. Once a channel is closed, any attempt to * invoke an I/O operation upon it will cause a {@link ClosedChannelException} * to be thrown. Whether or not a channel is open may be tested by invoking * its {@link #isOpen isOpen} method. * * <p> Channels are, in general, intended to be safe for multithreaded access * as described in the specifications of the interfaces and classes that extend * and implement this interface. * * * @author Mark Reinhold * @author JSR-51 Expert Group * @since 1.4 */public interface Channel extends Closeable { /** * Tells whether or not this channel is open. * * @return <tt>true</tt> if, and only if, this channel is open */ public boolean isOpen(); /** * Closes this channel. * * <p> After a channel is closed, any further attempt to invoke I/O * operations upon it will cause a {@link ClosedChannelException} to be * thrown. * * <p> If this channel is already closed then invoking this method has no * effect. * * <p> This method may be invoked at any time. If some other thread has * already invoked it, however, then another invocation will block until * the first invocation is complete, after which it will return without * effect. </p> * * @throws IOException If an I/O error occurs */ public void close() throws IOException;}

常见的channel实现有：

• FileChannel：文件读写数据通道
• SocketChannel：TCP读写网络数据通道
• ServerSocketChannel：服务端网络数据读写通道，可以监听TCP连接。对每一个新进来的连接都会创建一个SocketChannel。
• DatagramChannel：UDP读写网络数据通道

FileChannel

FileChannel是一个抽象类，它继承了AbstractInterruptibleChannel类，并实现了 SeekableByteChannel, GatheringByteChannel, ScatteringByteChannel接口。

具体的实现类主要是elImpl。下面详细分析下FileChannelImpl中每个方法的具体实现。

open

private FileChannelImpl(FileDescriptor var1, String var2, boolean var3, boolean var4, boolean var5, Object var6) { //主要记载操作系统维护的文件描述符  = var1;  //是否可读  = var3;  //是否可写  = var4;  //是否以追加的方式打开  = var5;   = var6;  = var2;  //底层使用native的read和write来处理文件的  = new FileDispatcherImpl(var5);}//FileInputStream::getChannel 调用 (fd, path, true, false, this) 获取只读channelpublic static FileChannel open(FileDescriptor var0, String var1, boolean var2, boolean var3, Object var4) { return new FileChannelImpl(var0, var1, var2, var3, false, var4);}//FileOutputStream::getChannel 调用 (fd, path, false, true, append, this) 获取只写channelpublic static FileChannel open(FileDescriptor var0, String var1, boolean var2, boolean var3, boolean var4, Object var5) { return new FileChannelImpl(var0, var1, var2, var3, var4, var5);}private FileChannelImpl(FileDescriptor fd, String path, boolean readable, boolean writable, boolean direct, Object parent){  = fd;  //是否可读  = readable;  //是否可写  = writable;  //对于从流创建的channel，在结束时要做不同的清理动作，（openJDK中才有，sun的jdk中没有）  = parent;  //源文件的path  = path;  //是否使用DirectIO  = direct;   = new FileDispatcherImpl(); if (direct) { assert path != null;  = IO(fd, path); } else {  = -1; } //当parent不存在时，则注册一个cleaner，否则交由parent做清理动作。 // Register a cleaning action if and only if there is no parent // as the parent will take care of closing the file descriptor. // FileChannel is used by the LambdaMetaFactory so a lambda cannot // be used here hence we use a nested class instead.  = parent != null ? null : ().register(this, new Closer(fd));}// Used by l(), l// and l()public static FileChannel open(FileDescriptor fd, String path, boolean readable, boolean writable, boolean direct, Object parent){ return new FileChannelImpl(fd, path, readable, writable, direct, parent);}

• open方法主要是返回一个新new的FileChannelImpl对象，初始化时设置fileDescriptor、readable、writable、append、parent、path等属性，看变量名很容易理解，在此不赘述变量含义。

read

//实现自SeekableByteChannel接口的方法，将文件中的内容读取到给定的byteBuffer中public int read(ByteBuffer dst) throws IOException { //保证读写时，channel处于开启状态 ensureOpen();  //判断是否可读 if (!readable) throw new NonReadableChannelException(); synchronized (positionLock) { if (direct) nelPositionAligned(position(), alignment); int n = 0; int ti = -1; try {  //开始阻塞，并注册为Interruptible，可以被中断 beginBlocking();  //将当前线程添加到NativeThreadSet中，并返回索引，方便后续操作。 //NativeThreadSet是一个线程安全的本地线程集合，方便管理，用来发送信号 ti = (); if (!isOpen()) return 0; do { //当未被系统中断（即读取完毕）或channel未被关闭，则一直读，将内容写入到byteBuffer（dst）中 n = (fd, dst, -1, direct, alignment, nd); } while ((n == ED) && isOpen()); return (n); } finally { //把当前线程从set中移出 (ti);  //结束，释放锁 endBlocking(n > 0); assert (n); } }}//实现自ScatteringByteChannel接口的方法，将文件中的内容依次读取到给定的byteBuffer数组中。public long read(ByteBuffer[] dsts, int offset, int length) throws IOException{ if ((offset < 0) || (length < 0) || (offset >  - length)) throw new IndexOutOfBoundsException(); //保证读写时，channel处于开启状态 ensureOpen();  //判断是否可读 if (!readable) throw new NonReadableChannelException(); synchronized (positionLock) { if (direct) nelPositionAligned(position(), alignment); long n = 0; int ti = -1; try { //开始阻塞，并注册为Interruptible，可以被中断 beginBlocking();  //将当前线程添加到NativeThreadSet中，并返回索引，方便后续操作。 //NativeThreadSet是一个线程安全的本地线程集合，方便管理，用来发送信号 ti = (); if (!isOpen()) return 0; do { //当未被系统中断（即读取完毕）或channel未被关闭，则一直读，将内容写入到byteBuffer（dst）中 n = (fd, dsts, offset, length, direct, alignment, nd); } while ((n == ED) && isOpen()); return (n); } finally { //把当前线程从set中移出 (ti);  //结束，释放锁 endBlocking(n > 0); assert (n); } }}

write

//实现自SeekableByteChannel接口的方法，将byteBuffer中的内容写入到文件中public int write(ByteBuffer src) throws IOException { //保证写时，channel处于开启状态 ensureOpen();  //判断是否可写 if (!writable) throw new NonWritableChannelException(); synchronized (positionLock) { if (direct) nelPositionAligned(position(), alignment); int n = 0; int ti = -1; try { //开始阻塞，并注册为Interruptible，可以被中断 beginBlocking();  //将当前线程添加到NativeThreadSet中，并返回索引，方便后续操作。 //NativeThreadSet是一个线程安全的本地线程集合，方便管理，用来发送信号 ti = (); if (!isOpen()) return 0; do { //当未被系统中断（即写入完毕）或channel未被关闭，则一直写，将内容写入到文件中 n = (fd, src, -1, direct, alignment, nd); } while ((n == ED) && isOpen()); return (n); } finally { //把当前线程从set中移出 (ti);  //结束，释放锁 assert (n); } }}//实现自GatheringByteChannel接口的方法，将byteBuffer数组中的内容依次写入到文件中public long write(ByteBuffer[] srcs, int offset, int length) throws IOException{ if ((offset < 0) || (length < 0) || (offset >  - length)) throw new IndexOutOfBoundsException(); //保证写时，channel处于开启状态 ensureOpen();  //判断是否可写 if (!writable) throw new NonWritableChannelException(); synchronized (positionLock) { if (direct) nelPositionAligned(position(), alignment); long n = 0; int ti = -1; try { //开始阻塞，并注册为Interruptible，可以被中断 beginBlocking();  //将当前线程添加到NativeThreadSet中，并返回索引，方便后续操作。 //NativeThreadSet是一个线程安全的本地线程集合，方便管理，用来发送信号 ti = (); if (!isOpen()) return 0; do { //当未被系统中断（即写入完毕）或channel未被关闭，则一直写，将内容写入到文件中 n = (fd, srcs, offset, length, direct, alignment, nd); } while ((n == ED) && isOpen()); return (n); } finally { //把当前线程从set中移出 (ti);  //结束，释放锁 assert (n); } }}

position

//实现自SeekableByteChannel接口的方法，获取当前channel的positionpublic long position() throws IOException { ensureOpen(); synchronized (positionLock) { long p = -1; int ti = -1; try { beginBlocking(); ti = (); if (!isOpen()) return 0; boolean append = (fd); do { //append模式下，position在channel的末尾 // in append-mode then position is advanced to end before writing p = (append) ? (fd) : (fd, -1); } while ((p == ED) && isOpen()); return (p); } finally { (ti); endBlocking(p > -1); assert (p); } }}//实现自SeekableByteChannel接口的方法，设置当前channel的position为newPositionpublic FileChannel position(long newPosition) throws IOException { ensureOpen(); if (newPosition < 0) throw new IllegalArgumentException(); synchronized (positionLock) { long p = -1; int ti = -1; try { beginBlocking(); ti = (); if (!isOpen()) return null; do { //设置当前position为newPosition p = (fd, newPosition); } while ((p == ED) && isOpen()); return this; } finally { (ti); endBlocking(p > -1); assert (p); } }}

size

实现自SeekableByteChannel接口的方法，返回当前实体（文件）的大小

truncate

实现自SeekableByteChannel接口的方法，用来截取文件至newSize大小

force

实现自SeekableByteChannel接口的方法，用来将channel中尚未写入磁盘的数据强制落盘

transferTo

将fileChannel中的数据传递至另一个channel

transferFrom

从其它channel读取数据至fileChannel

SocketChannel

open

/** * Opens a socket channel. * * <p> The new channel is created by invoking the {@link * rovider#openSocketChannel * openSocketChannel} method of the system-wide default {@link * rovider} object. </p> * * @return A new socket channel * * @throws IOException * If an I/O error occurs */public static SocketChannel open() throws IOException { return ().openSocketChannel();}

open方法是调用SelectorProvider中实现了rovider#openSocketChannel的方法，底层实际是new SocketChannelImpl，调用native方法创建socket

connect

public boolean connect(SocketAddress sa) throws IOException { //校验Address是否合法 InetSocketAddress isa = ess(sa);  //获取系统安全管理器 SecurityManager sm = tyManager(); if (sm != null)  //校验IP和端口是否被允许连接 ect(s().getHostAddress(), ()); InetAddress ia = s();  //如果是本机地址，则获取本机的host if (lAddress()) ia = ost(); try { //加读锁 (); try { //加写锁 (); try { int n = 0;  //是否阻塞 boolean blocking = isBlocking(); try { //开启connect前的校验并设置为ST_CONNECTIONPENDING，如果blocking是true 即阻塞模式，则记录当前线程的ID，以便接收信号处理。 beginConnect(blocking, isa); do { //调用native connect方法 n = (fd, ia, ()); } while (n == ED && isOpen()); } finally { //结束连接 endConnect(blocking, (n > 0)); } assert (n); return n > 0; } finally { //释放写锁 (); } } finally { //释放读锁 (); } } catch (IOException ioe) { // connect failed, close the channel close(); throw (ioe, isa); }}

configureBlocking

实现自SelectableChannel的接口方法，调用native方法设置socket的阻塞状态

register

在AbstractSelectableChannel中定义，注册要监听的事件。

public final SelectionKey register(Selector sel, int ops, Object att) throws ClosedChannelException{ if ((ops & ~validOps()) != 0) throw new IllegalArgumentException(); if (!isOpen()) throw new ClosedChannelException(); synchronized (regLock) { if (isBlocking()) throw new IllegalBlockingModeException(); synchronized (keyLock) { // re-check if channel has been closed if (!isOpen()) throw new ClosedChannelException(); SelectionKey k = findKey(sel); if (k != null) { (att); ps(ops); } else { // 向Selector中注册事件 // New registration k = ((AbstractSelector)sel).register(this, ops, att); addKey(k); } return k; } }}

read

//实现自ReadableByteChannel接口的方法，从socket中读取数据至ByteBuffer@Overridepublic int read(ByteBuffer buf) throws IOException { nNull(buf); (); try { boolean blocking = isBlocking(); int n = 0; try { //检查channel是否开启并已经是connected的状态。如果blocking是true 即阻塞模式，则记录当前线程的ID，以便接收信号处理。 beginRead(blocking); // check if input is shutdown if (isInputClosed) return ; //如果是阻塞模式，则一直读取直到数据读取完毕；非阻塞模式则直接调用native方法不需要等待。 if (blocking) { do { n = (fd, buf, -1, nd); } while (n == ED && isOpen()); } else { n = (fd, buf, -1, nd); } } finally { endRead(blocking, n > 0); if (n <= 0 && isInputClosed) return ; } return (n); } finally { (); }}//实现自ScatteringByteChannel接口的方法，从socket中依次读取数据至ByteBuffer数组@Overridepublic long read(ByteBuffer[] dsts, int offset, int length) throws IOException{ IndexSize(offset, length, ); (); try { boolean blocking = isBlocking(); long n = 0; try { beginRead(blocking); // check if input is shutdown if (isInputClosed) return ; //如果是阻塞模式，则一直读取直到数据读取完毕；非阻塞模式则直接调用native方法不需要等待。 if (blocking) { do { n = (fd, dsts, offset, length, nd); } while (n == ED && isOpen()); } else { n = (fd, dsts, offset, length, nd); } } finally { endRead(blocking, n > 0); if (n <= 0 && isInputClosed) return ; } return (n); } finally { (); }}

write

//实现自ReadableByteChannel接口的方法，将ByteBuffer中的数据写入socket@Overridepublic int write(ByteBuffer buf) throws IOException { nNull(buf); (); try { boolean blocking = isBlocking(); int n = 0; try { beginWrite(blocking); //如果是阻塞模式，则一直读取直到数据读取完毕；非阻塞模式则直接调用native方法不需要等待。 if (blocking) { do { n = (fd, buf, -1, nd); } while (n == ED && isOpen()); } else { n = (fd, buf, -1, nd); } } finally { endWrite(blocking, n > 0); if (n <= 0 && isOutputClosed) throw new AsynchronousCloseException(); } return (n); } finally { (); }}@Overridepublic long write(ByteBuffer[] srcs, int offset, int length) throws IOException{ IndexSize(offset, length, ); (); try { boolean blocking = isBlocking(); long n = 0; try { beginWrite(blocking); //如果是阻塞模式，则一直等待直到数据写入完毕；非阻塞模式则直接调用native方法不需要等待。 if (blocking) { do { n = (fd, srcs, offset, length, nd); } while (n == ED && isOpen()); } else { n = (fd, srcs, offset, length, nd); } } finally { endWrite(blocking, n > 0); if (n <= 0 && isOutputClosed) throw new AsynchronousCloseException(); } return (n); } finally { (); }}//实现自ReadableByteChannel接口的方法，将ByteBuffer数组中的数据依次写入socket/** * Writes a byte of out of band data. */int sendOutOfBandData(byte b) throws IOException { (); try { boolean blocking = isBlocking(); int n = 0; try { beginWrite(blocking); //如果是阻塞模式，则一直等待直到数据写入完毕；非阻塞模式则直接调用native方法不需要等待。 if (blocking) { do { n = sendOutOfBandData(fd, b); } while (n == ED && isOpen()); } else { n = sendOutOfBandData(fd, b); } } finally { endWrite(blocking, n > 0); if (n <= 0 && isOutputClosed) throw new AsynchronousCloseException(); } return (n); } finally { (); }}

ServerSocketChannel

socket

@Overridepublic ServerSocket socket() { synchronized (stateLock) { if (socket == null) socket = (this); return socket; }}

bind

@Overridepublic ServerSocketChannel bind(SocketAddress local, int backlog) throws IOException { synchronized (stateLock) { ensureOpen(); if (localAddress != null) throw new AlreadyBoundException(); InetSocketAddress isa = (local == null) ? new InetSocketAddress(0) : ess(local); SecurityManager sm = tyManager(); if (sm != null) en(());  //绑定前做一些前置处理，如将tcp socket文件描述符转换成SDP Bind(fd, s(), ());  //绑定IP和地址 (fd, s(), ());  //开始监听，设置socket上最多可以挂起backlog个连接，若backlog小于1 则默认设置50个 (fd, backlog < 1 ? 50 : backlog);  localAddress = ess(fd); } return this;}

accept

@Overridepublic SocketChannel accept() throws IOException { (); try { int n = 0; FileDescriptor newfd = new FileDescriptor(); InetSocketAddress[] isaa = new InetSocketAddress[1]; boolean blocking = isBlocking(); try { begin(blocking); do { //阻塞等待接收客户端链接 n = accept(, newfd, isaa); } while (n == ED && isOpen()); } finally { end(blocking, n > 0); assert (n); } if (n < 1) return null; //新接收的socket初始设置为阻塞模式（因此非阻塞模式的每次需要显示设置） // newly accepted socket is initially in blocking mode Blocking(newfd, true); InetSocketAddress isa = isaa[0];  //用新接收的socket创建SocketChannel SocketChannel sc = new SocketChannelImpl(provider(), newfd, isa); // check permitted to accept connections from the remote address SecurityManager sm = tyManager(); if (sm != null) { try { pt(s().getHostAddress(), ()); } catch (SecurityException x) { (); throw x; } } return sc; } finally { (); }}

ServerSocketChannel并没有read和write方法，只是继承了AbstractSelectableChannel，以便在selector中使用

DatagramChannel

open

public DatagramChannelImpl(SelectorProvider sp) throws IOException{ super(sp); Create(); try { //如果不支持IPv6则使用IPv4  = 6Available() ? 6 : ;  //设置非流式的socket（tcp是流模式协议，udp是数据报模式协议）  = (family, false);  = (fd); } catch (IOException ioe) { lose(); throw ioe; }}

receive

public SocketAddress receive(ByteBuffer dst) throws IOException { if (y()) throw new IllegalArgumentException("Read-only buffer"); (); try { boolean blocking = isBlocking(); int n = 0; ByteBuffer bb = null; try { SocketAddress remote = beginRead(blocking, false); boolean connected = (remote != null); SecurityManager sm = tyManager(); if (connected || (sm == null)) { // connected or no security manager do { n = receive(fd, dst, connected); } while ((n == ED) && isOpen()); if (n == LE) return null; } else { // Cannot receive into user's buffer when running with a // security manager and not connected bb = aryDirectBuffer(()); for (;;) { do { n = receive(fd, bb, connected); } while ((n == ED) && isOpen()); if (n == LE) return null; InetSocketAddress isa = (InetSocketAddress)sender; try { pt(s().getHostAddress(), ()); } catch (SecurityException se) { // Ignore packet (); n = 0; continue; } (); (bb); break; } }  //sender:发送方地址， Set by receive0 (## ugh) assert sender != null; return sender; } finally { if (bb != null) mporaryDirectBuffer(bb); endRead(blocking, n > 0); assert (n); } } finally { (); }}

send

public int send(ByteBuffer src, SocketAddress target) throws IOException{ nNull(src); InetSocketAddress isa = ess(target, family); (); try { boolean blocking = isBlocking(); int n = 0; try { //当connect后，remote会设置为连接的地址 SocketAddress remote = beginWrite(blocking, false); if (remote != null) { // connected if (!(remote)) { throw new AlreadyConnectedException(); } do { n = (fd, src, -1, nd); } while ((n == ED) && isOpen()); } else { // not connected SecurityManager sm = tyManager(); if (sm != null) { InetAddress ia = s(); if (stAddress()) { icast(ia); } else { ect(dress(), ()); } } do { n = send(fd, src, isa); } while ((n == ED) && isOpen()); } } finally { endWrite(blocking, n > 0); assert (n); } return (n); } finally { (); }}

connect

@Overridepublic DatagramChannel connect(SocketAddress sa) throws IOException { InetSocketAddress isa = ess(sa, family); SecurityManager sm = tyManager(); if (sm != null) { InetAddress ia = s(); if (stAddress()) { icast(ia); } else { ect(dress(), ()); pt(dress(), ()); } } (); try { (); try { synchronized (stateLock) { ensureOpen(); if (state == ST_CONNECTED) throw new AlreadyConnectedException(); int n = (family, fd, s(), ()); if (n <= 0) throw new Error(); // Can't happen // connected remoteAddress = isa; state = ST_CONNECTED; // refresh local address localAddress = ess(fd); // flush any packets already received. boolean blocking = isBlocking(); if (blocking) { Blocking(fd, false); } try { ByteBuffer buf = (100); while (receive(buf) != null) { (); } } finally { if (blocking) { Blocking(fd, true); } } } } finally { (); } } finally { (); } return this;}

udp是数据报模式的协议，是没有connect的。这里的connect实际上是在底层忽略了与其他地址的数据传输。

在connect后，就可以像socketChannel似得使用read和write了

总结

本文学习了各种channel的实现，主要是对底层native方法的一些封装，针对不同属性的实体（文件、socket），使用对应的channel与byteBuffer传输数据。再通过byteBuffer与byte数据进行转换。

channel的实现中，封装了大量的native方法，重要的底层实现全在native中，后续可以深入学习下。

本文中出现的byteBuffer和selector将在接下来的文章中，单独分析。

作者：aloof_