jdk源码解析八之NIO(socketChannel)

亦凉 2023-02-24 03:59 81阅读 0赞

ServerSocketChannel

  1. ServerSocketChannelImpl(SelectorProvider sp) throws IOException {
  3.         super(sp);
  4.         //创建FileDescriptor
  5.         this.fd =  Net.serverSocket(true);
  6.         //获取到文件描述符的值。
  7.         this.fdVal = IOUtil.fdVal(fd);
  8.         //标记状态为使用
  9.         this.state = ST_INUSE;
  10.     }
  12.     ServerSocketChannelImpl(SelectorProvider sp,
  13.                             FileDescriptor fd,
  14.                             boolean bound)
  15.         throws IOException
  16.     {
  18.         super(sp);
  19.         this.fd =  fd;
  20.         this.fdVal = IOUtil.fdVal(fd);
  21.         this.state = ST_INUSE;
  22.         //已绑定则直接获取地址
  23.         if (bound)
  24.             //获取传入的文件描述符的socket地址
  25.             localAddress = Net.localAddress(fd);
  26.     }
  28.   @Override
  29.     public ServerSocketChannel bind(SocketAddress local, int backlog) throws IOException {
  31.         synchronized (lock) {
  33.             //判断是否通道打开以及是否绑定
  34.             if (!isOpen())
  35.                 throw new ClosedChannelException();
  36.             if (isBound())
  37.                 throw new AlreadyBoundException();
  38.             //绑定地址为null处理
  39.             InetSocketAddress isa = (local == null) ? new InetSocketAddress(0) :
  40.                 Net.checkAddress(local);
  41.             SecurityManager sm = System.getSecurityManager();
  42.             //检测端口是否被监听
  43.             if (sm != null)
  44.                 sm.checkListen(isa.getPort());
  45.             //将fd转换为SDP(Sockets Direct Protocol,Java套接字直接协议) socket。
  46.             //SDP需要网卡支持InfiniBand高速网络通信技术，windows不支持该协议。
  47.             NetHooks.beforeTcpBind(fd, isa.getAddress(), isa.getPort());
  48.             //连接
  49.             Net.bind(fd, isa.getAddress(), isa.getPort());
  50.             //默认tcp待连接队列长度为50
  51.             Net.listen(fd, backlog < 1 ? 50 : backlog);
  52.             synchronized (stateLock) {
  54.                 //获取文件描述符的socket地址
  55.                 localAddress = Net.localAddress(fd);
  56.             }
  57.         }
  58.         return this;
  59.     }
  61.     public SocketChannel accept() throws IOException {
  63.         synchronized (lock) {
  65.             if (!isOpen())
  66.                 throw new ClosedChannelException();
  67.             if (!isBound())
  68.                 throw new NotYetBoundException();
  69.             SocketChannel sc = null;
  71.             int n = 0;
  72.             FileDescriptor newfd = new FileDescriptor();
  73.             InetSocketAddress[] isaa = new InetSocketAddress[1];
  75.             try {
  77.                 begin();
  78.                 if (!isOpen())
  79.                     return null;
  80.                 thread = NativeThread.current();
  81.                 for (;;) {
  83.                     //接收一个新的连接
  84.                     //将isaa设置为socket的远程地址
  85.                     //将给定的文件描述符与socket客户端绑定
  86.                     n = accept(this.fd, newfd, isaa);
  87.                     //返回1成功
  88.                     //系统执行中断,继续等待接收
  89.                     if ((n == IOStatus.INTERRUPTED) && isOpen())
  90.                         continue;
  91.                     break;
  92.                 }
  93.             } finally {
  95.                 thread = 0;
  96.                 end(n > 0);
  97.                 assert IOStatus.check(n);
  98.             }
  100.             //返回1成功
  101.             if (n < 1)
  102.                 return null;
  103.             //默认阻塞模式
  104.             IOUtil.configureBlocking(newfd, true);
  105.             InetSocketAddress isa = isaa[0];
  106.             //初始化客户端SocketChannel
  107.             sc = new SocketChannelImpl(provider(), newfd, isa);
  108.             //检查权限
  109.             SecurityManager sm = System.getSecurityManager();
  110.             if (sm != null) {
  112.                 try {
  114.                     sm.checkAccept(isa.getAddress().getHostAddress(),
  115.                                    isa.getPort());
  116.                 } catch (SecurityException x) {
  118.                     sc.close();
  119.                     throw x;
  120.                 }
  121.             }
  122.             return sc;
  124.         }
  125.     }
  127.     //配置是否阻塞
  128.     protected void implConfigureBlocking(boolean block) throws IOException {
  130.         IOUtil.configureBlocking(fd, block);
  131.     }
  133. //关闭socket
  134.     protected void implCloseSelectableChannel() throws IOException {
  136.         synchronized (stateLock) {
  138.             if (state != ST_KILLED)
  139.                 nd.preClose(fd);
  140.             long th = thread;
  141.             if (th != 0)
  142.                 //发送信号给线程，将其从阻塞I/O中释放，避免一直被阻塞。
  143.                 NativeThread.signal(th);
  144.             //若还有注册的channel，则不处理，等待key全部注销后再kill
  145.             //若没有的话可以直接kill当前channel
  146.             if (!isRegistered())
  147.                 kill();
  148.         }
  149.     }

SocketChannel

生命周期

在这里插入图片描述

connect

  1. public static SocketChannel open(SocketAddress remote)
  2.         throws IOException
  3.     {
  5.         SocketChannel sc = open();
  6.         try {
  8.             //建立连接
  9.             sc.connect(remote);
  10.         } catch (Throwable x) {
  12.             try {
  14.                 sc.close();
  15.             } catch (Throwable suppressed) {
  17.                 x.addSuppressed(suppressed);
  18.             }
  19.             throw x;
  20.         }
  21.         assert sc.isConnected();
  22.         return sc;
  23.     }
  25.   public boolean connect(SocketAddress sa) throws IOException {
  27.         int localPort = 0;
  29.         synchronized (readLock) {
  31.             synchronized (writeLock) {
  33.                 ensureOpenAndUnconnected();
  34.                 InetSocketAddress isa = Net.checkAddress(sa);
  35.                 SecurityManager sm = System.getSecurityManager();
  36.                 if (sm != null)
  37.                     sm.checkConnect(isa.getAddress().getHostAddress(),
  38.                                     isa.getPort());
  40.                 synchronized (blockingLock()) {
  42.                     int n = 0;
  43.                     try {
  45.                         try {
  47.                             begin();
  48.                             synchronized (stateLock) {
  50.                                 if (!isOpen()) {
  52.                                     return false;
  53.                                 }
  54.                                 // notify hook only if unbound
  55.                                 if (localAddress == null) {
  57.                                     //将fd转换为SDP(Sockets Direct Protocol,Java套接字直接协议) socket。
  58.                                     NetHooks.beforeTcpConnect(fd,
  59.                                                            isa.getAddress(),
  60.                                                            isa.getPort());
  61.                                 }
  62.                                 readerThread = NativeThread.current();
  63.                             }
  64.                             for (;;) {
  66.                                 InetAddress ia = isa.getAddress();
  67.                                 if (ia.isAnyLocalAddress())
  68.                                     //返回本地主机地址
  69.                                     ia = InetAddress.getLocalHost();
  70.                                 //建立连接
  71.                                 n = Net.connect(fd,
  72.                                                 ia,
  73.                                                 isa.getPort());
  74.                                 if (  (n == IOStatus.INTERRUPTED)
  75.                                       && isOpen())
  76.                                     continue;
  77.                                 break;
  78.                             }
  80.                         } finally {
  82.                             readerCleanup();
  83.                             end((n > 0) || (n == IOStatus.UNAVAILABLE));
  84.                             assert IOStatus.check(n);
  85.                         }
  86.                     } catch (IOException x) {
  88.                         // If an exception was thrown, close the channel after
  89.                         // invoking end() so as to avoid bogus
  90.                         // AsynchronousCloseExceptions
  91.                         close();
  92.                         throw x;
  93.                     }
  94.                     synchronized (stateLock) {
  96.                         remoteAddress = isa;
  97.                         if (n > 0) {
  100.                             // Connection succeeded; disallow further
  101.                             // invocation
  102.                             //更新状态
  103.                             state = ST_CONNECTED;
  104.                             if (isOpen())
  105.                                 localAddress = Net.localAddress(fd);
  106.                             return true;
  107.                         }
  108.                         // If nonblocking and no exception then connection
  109.                         // pending; disallow another invocation
  110.                         //非阻塞状态下更新为待连接
  111.                         if (!isBlocking())
  112.                             state = ST_PENDING;
  113.                         else
  114.                             assert false;
  115.                     }
  116.                 }
  117.                 return false;
  118.             }
  119.         }
  120.     }

关于linux的聚集写和散步读具体内容,参考readv()和writev()函数

write

  1. public int write(ByteBuffer buf) throws IOException {
  3.         if (buf == null)
  4.             throw new NullPointerException();
  5.         synchronized (writeLock) {
  7.             ensureWriteOpen();
  8.             int n = 0;
  9.             try {
  11.                 begin();
  12.                 synchronized (stateLock) {
  14.                     if (!isOpen())
  15.                         return 0;
  16.                     writerThread = NativeThread.current();
  17.                 }
  18.                 for (;;) {
  20.                     //写入buf
  21.                     n = IOUtil.write(fd, buf, -1, nd);
  22.                     if ((n == IOStatus.INTERRUPTED) && isOpen())
  23.                         continue;
  24.                     return IOStatus.normalize(n);
  25.                 }
  26.             } finally {
  28.                 writerCleanup();
  29.                 end(n > 0 || (n == IOStatus.UNAVAILABLE));
  30.                 synchronized (stateLock) {
  32.                     if ((n <= 0) && (!isOutputOpen))
  33.                         throw new AsynchronousCloseException();
  34.                 }
  35.                 assert IOStatus.check(n);
  36.             }
  37.         }
  38.     }
  40. static int write(FileDescriptor fd, ByteBuffer src, long position,
  41.                      NativeDispatcher nd)
  42.         throws IOException
  43.     {
  45.         //如果是直接缓冲区则写入本地缓冲区中
  46.         if (src instanceof DirectBuffer)
  47.             return writeFromNativeBuffer(fd, src, position, nd);
  49.         //如果缓冲区是基于堆的,会多一次内存复制
  50.         // Substitute a native buffer
  51.         int pos = src.position();
  52.         int lim = src.limit();
  53.         assert (pos <= lim);
  54.         int rem = (pos <= lim ? lim - pos : 0);
  55.         //创建临时缓冲区
  56.         ByteBuffer bb = Util.getTemporaryDirectBuffer(rem);
  57.         try {
  59.             //写入临时缓冲区中
  60.             bb.put(src);
  61.             bb.flip();
  62.             // Do not update src until we see how many bytes were written
  63.             src.position(pos);
  65.             //直接缓冲区数据写入
  66.             int n = writeFromNativeBuffer(fd, bb, position, nd);
  67.             if (n > 0) {
  69.                 // now update src
  70.                 //更新实际写入量
  71.                 src.position(pos + n);
  72.             }
  73.             return n;
  74.         } finally {
  76.             //释放临时缓冲区
  77.             Util.offerFirstTemporaryDirectBuffer(bb);
  78.         }
  79.     }

当写入的buffer使用堆,会多一次内存复制.《Java NIO为什么需要DirectByteBuffer作为中间缓冲区》

  1. private static ThreadLocal<BufferCache> bufferCache =
  2.         new ThreadLocal<BufferCache>()
  3.     {
  5.         @Override
  6.         protected BufferCache initialValue() {
  8.             return new BufferCache();
  9.         }
  10.     };
  12.  public static ByteBuffer getTemporaryDirectBuffer(int size) {
  14.          //如果所需缓冲区太大,直接新建
  15.         if (isBufferTooLarge(size)) {
  17.             return ByteBuffer.allocateDirect(size);
  18.         }
  20.         //从线程缓冲区中,获取一个缓冲区
  21.         BufferCache cache = bufferCache.get();
  22.         //获取容量>=size的缓冲区
  23.         ByteBuffer buf = cache.get(size);
  24.         if (buf != null) {
  26.             return buf;
  27.         } else {
  29.             //当没有可用缓冲区
  30.             if (!cache.isEmpty()) {
  32.                 //第一个缓冲区移除缓存
  33.                 buf = cache.removeFirst();
  34.                 //释放
  35.                 free(buf);
  36.             }
  37.             //重新分配一个合适大小的缓冲区
  38.             return ByteBuffer.allocateDirect(size);
  39.         }
  40.     }
  41.         ByteBuffer get(int size) {
  43.             // Don't call this if the buffer would be too large.
  44.             //断言缓冲区是否过大
  45.             assert !isBufferTooLarge(size);
  47.             //缓存没直接返回
  48.             if (count == 0)
  49.                 return null;  // cache is empty
  51.             ByteBuffer[] buffers = this.buffers;
  53.             // search for suitable buffer (often the first buffer will do)
  54.             ByteBuffer buf = buffers[start];
  55.             //如果取出第一个buf容量<size,则遍历查找
  56.             if (buf.capacity() < size) {
  58.                 buf = null;
  59.                 int i = start;
  60.                 //查找>=size缓冲区
  61.                 while ((i = next(i)) != start) {
  63.                     ByteBuffer bb = buffers[i];
  64.                     if (bb == null)
  65.                         break;
  66.                     if (bb.capacity() >= size) {
  68.                         buf = bb;
  69.                         break;
  70.                     }
  71.                 }
  72.                 if (buf == null)
  73.                     return null;
  74.                 // move first element to here to avoid re-packing
  75.                 //覆盖取出的元素
  76.                 buffers[i] = buffers[start];
  77.             }
  79.             // remove first element
  80.             //start下标的buf不管满足size与否,都会置空,如果满足size,置空后start移动到下一个位置
  81.             //如果不满足size,则从后遍历查找到符合size要求的,然后start覆盖,start移动到下一个位置
  82.             buffers[start] = null;
  83.             start = next(start);
  84.             count--;
  86.             // prepare the buffer and return it
  87.             //重置pos,设置limit上限
  88.             buf.rewind();
  89.             buf.limit(size);
  90.             return buf;
  91.         }
  94.   static void offerFirstTemporaryDirectBuffer(ByteBuffer buf) {
  96.         // If the buffer is too large for the cache we don't have to
  97.         // check the cache. We'll just free it.
  98.         //太大直接释放
  99.         if (isBufferTooLarge(buf)) {
  101.             free(buf);
  102.             return;
  103.         }
  105.         assert buf != null;
  106.         BufferCache cache = bufferCache.get();
  107.         //
  108.         if (!cache.offerFirst(buf)) {
  110.             // cache is full
  111.             //满了直接释放
  112.             free(buf);
  113.         }
  114.     }
  116.   boolean offerFirst(ByteBuffer buf) {
  118.             // Don't call this if the buffer is too large.
  119.             //断言
  120.             assert !isBufferTooLarge(buf);
  122.             //超过缓冲容量,返回false
  123.             if (count >= TEMP_BUF_POOL_SIZE) {
  125.                 return false;
  126.             } else {
  128.                 //计算前一个位置,前一位绝对是为null的
  129.                 start = (start + TEMP_BUF_POOL_SIZE - 1) % TEMP_BUF_POOL_SIZE;
  130.                 buffers[start] = buf;
  131.                 count++;
  132.                 return true;
  133.             }
  134.         }

聚集写

按照顺序写入多个buf

  1. //一次性写入多个buf
  2.     //聚集写
  3.     public long write(ByteBuffer[] srcs, int offset, int length)
  4.         throws IOException
  5.     {
  7.         if ((offset < 0) || (length < 0) || (offset > srcs.length - length))
  8.             throw new IndexOutOfBoundsException();
  9.         synchronized (writeLock) {
  11.             ensureWriteOpen();
  12.             long n = 0;
  13.             try {
  15.                 begin();
  16.                 synchronized (stateLock) {
  18.                     if (!isOpen())
  19.                         return 0;
  20.                     writerThread = NativeThread.current();
  21.                 }
  22.                 for (;;) {
  24.                     //写入多个buf
  25.                     n = IOUtil.write(fd, srcs, offset, length, nd);
  26.                     if ((n == IOStatus.INTERRUPTED) && isOpen())
  27.                         continue;
  28.                     return IOStatus.normalize(n);
  29.                 }
  30.             } finally {
  32.                 writerCleanup();
  33.                 end((n > 0) || (n == IOStatus.UNAVAILABLE));
  34.                 synchronized (stateLock) {
  36.                     if ((n <= 0) && (!isOutputOpen))
  37.                         throw new AsynchronousCloseException();
  38.                 }
  39.                 assert IOStatus.check(n);
  40.             }
  41.         }
  42.     }
  44. public class IOUtil {
  46.  static long write(FileDescriptor fd, ByteBuffer[] bufs, int offset, int length,
  47.                       NativeDispatcher nd)
  48.         throws IOException
  49.     {
  51.         //封装多个buf
  52.         IOVecWrapper vec = IOVecWrapper.get(length);
  54.         boolean completed = false;
  55.         int iov_len = 0;
  56.         try {
  59.             // Iterate over buffers to populate native iovec array.
  60.             int count = offset + length;
  61.             int i = offset;
  62.             while (i < count && iov_len < IOV_MAX) {
  64.                 //遍历每一块缓冲区
  65.                 ByteBuffer buf = bufs[i];
  66.                 //计算可读大小
  67.                 int pos = buf.position();
  68.                 int lim = buf.limit();
  69.                 assert (pos <= lim);
  70.                 int rem = (pos <= lim ? lim - pos : 0);
  72.                 //将buf放入IOVecWrapper
  73.                 if (rem > 0) {
  75.                     vec.setBuffer(iov_len, buf, pos, rem);
  77.                     // allocate shadow buffer to ensure I/O is done with direct buffer
  78.                     //针对buf不是直接缓冲区的处理
  79.                     if (!(buf instanceof DirectBuffer)) {
  81.                         ByteBuffer shadow = Util.getTemporaryDirectBuffer(rem);
  82.                         shadow.put(buf);
  83.                         shadow.flip();
  84.                         vec.setShadow(iov_len, shadow);
  85.                         buf.position(pos);  // temporarily restore position in user buffer
  86.                         buf = shadow;
  87.                         pos = shadow.position();
  88.                     }
  90.                     //设置缓冲区的起始地址
  91.                     vec.putBase(iov_len, ((DirectBuffer)buf).address() + pos);
  92.                     //设置缓冲区长度
  93.                     vec.putLen(iov_len, rem);
  94.                     iov_len++;
  95.                 }
  96.                 i++;
  97.             }
  98.             if (iov_len == 0)
  99.                 return 0L;
  101.             //写入多个缓冲区数据
  102.             long bytesWritten = nd.writev(fd, vec.address, iov_len);
  104.             // Notify the buffers how many bytes were taken
  105.             long left = bytesWritten;
  106.             for (int j=0; j<iov_len; j++) {
  108.                 if (left > 0) {
  110.                     ByteBuffer buf = vec.getBuffer(j);
  111.                     int pos = vec.getPosition(j);
  112.                     int rem = vec.getRemaining(j);
  113.                     int n = (left > rem) ? rem : (int)left;
  114.                     buf.position(pos + n);
  115.                     left -= n;
  116.                 }
  117.                 // return shadow buffers to buffer pool
  118.                 ByteBuffer shadow = vec.getShadow(j);
  119.                 if (shadow != null)
  120.                     //将写入的缓冲放入临时直接缓冲区
  121.                     Util.offerLastTemporaryDirectBuffer(shadow);
  122.                 //清除缓存
  123.                 vec.clearRefs(j);
  124.             }
  126.             completed = true;
  127.             return bytesWritten;
  129.         } finally {
  131.             // if an error occurred then clear refs to buffers and return any shadow
  132.             // buffers to cache
  133.             if (!completed) {
  135.                 for (int j=0; j<iov_len; j++) {
  137.                     ByteBuffer shadow = vec.getShadow(j);
  138.                     if (shadow != null)
  139.                         Util.offerLastTemporaryDirectBuffer(shadow);
  140.                     vec.clearRefs(j);
  141.                 }
  142.             }
  143.         }
  144.     }
  146. }
  148. class IOVecWrapper {
  150.     // per thread IOVecWrapper
  151.     //每个线程保存一份IOVecWrapper缓存
  152.     private static final ThreadLocal<IOVecWrapper> cached =
  153.         new ThreadLocal<IOVecWrapper>();
  155.     private IOVecWrapper(int size) {
  157.         this.size      = size;
  158.         this.buf       = new ByteBuffer[size];
  159.         this.position  = new int[size];
  160.         this.remaining = new int[size];
  161.         this.shadow    = new ByteBuffer[size];
  162.         //false:无需页面对齐
  163.         this.vecArray  = new AllocatedNativeObject(size * SIZE_IOVEC, false);
  164.         this.address   = vecArray.address();
  165.     }
  167.     //通过get获取一块适合大小的空间
  168.     static IOVecWrapper get(int size) {
  170.         IOVecWrapper wrapper = cached.get();
  171.         if (wrapper != null && wrapper.size < size) {
  173.             // not big enough; eagerly release memory
  174.             //若获取到空间不够大，则重新初始化一个空间。
  175.             wrapper.vecArray.free();
  176.             wrapper = null;
  177.         }
  178.         if (wrapper == null) {
  180.             wrapper = new IOVecWrapper(size);
  181.             //native资源，当对象释放时使得操作系统可以释放内存
  182.             Cleaner.create(wrapper, new Deallocator(wrapper.vecArray));
  183.             cached.set(wrapper);
  184.         }
  185.         return wrapper;
  186.     }
  187.     //保存buf
  188.     void setBuffer(int i, ByteBuffer buf, int pos, int rem) {
  190.         this.buf[i] = buf;
  191.         this.position[i] = pos;
  192.         this.remaining[i] = rem;
  193.     }
  195.    static {
  197.         //获取本机指针大小
  198.         addressSize = Util.unsafe().addressSize();
  199.         //保存每个指针偏移量
  200.         LEN_OFFSET = addressSize;
  201.         //用于保存每个AllocatedNativeObject对象的元素的大小
  202.         //每个NativeObject有两个long属性，因此需要×2
  203.         SIZE_IOVEC = (short) (addressSize * 2);
  204.     }
  205.     }

read

读的时候，如果采用的时堆缓冲区，会多一遍读取到直接缓冲区,然后在读取到堆缓冲区

  1. static int read(FileDescriptor fd, ByteBuffer dst, long position,
  2.                     NativeDispatcher nd)
  3.         throws IOException
  4.     {
  6.         if (dst.isReadOnly())
  7.             throw new IllegalArgumentException("Read-only buffer");
  8.         if (dst instanceof DirectBuffer)
  9.             //使用直接缓冲区
  10.             return readIntoNativeBuffer(fd, dst, position, nd);
  12.         // Substitute a native buffer
  13.         //获取临时直接缓冲区
  14.         ByteBuffer bb = Util.getTemporaryDirectBuffer(dst.remaining());
  15.         try {
  17.             int n = readIntoNativeBuffer(fd, bb, position, nd);
  18.             bb.flip();
  19.             //临时缓冲区数据写入堆内存中
  20.             if (n > 0)
  21.                 dst.put(bb);
  22.             return n;
  23.         } finally {
  25.             //释放临时缓冲区
  26.             Util.offerFirstTemporaryDirectBuffer(bb);
  27.         }
  28.     }

分散读

  1. static long read(FileDescriptor fd, ByteBuffer[] bufs, int offset, int length,
  2.                      NativeDispatcher nd)
  3.         throws IOException
  4.     {
  6.         IOVecWrapper vec = IOVecWrapper.get(length);
  8.         boolean completed = false;
  9.         int iov_len = 0;
  10.         try {
  13.             // Iterate over buffers to populate native iovec array.
  14.             int count = offset + length;
  15.             int i = offset;
  16.             while (i < count && iov_len < IOV_MAX) {
  18.                 ByteBuffer buf = bufs[i];
  19.                 if (buf.isReadOnly())
  20.                     throw new IllegalArgumentException("Read-only buffer");
  21.                 int pos = buf.position();
  22.                 int lim = buf.limit();
  23.                 assert (pos <= lim);
  24.                 int rem = (pos <= lim ? lim - pos : 0);
  26.                 if (rem > 0) {
  28.                     vec.setBuffer(iov_len, buf, pos, rem);
  30.                     // allocate shadow buffer to ensure I/O is done with direct buffer
  31.                     if (!(buf instanceof DirectBuffer)) {
  33.                         ByteBuffer shadow = Util.getTemporaryDirectBuffer(rem);
  34.                         vec.setShadow(iov_len, shadow);
  35.                         buf = shadow;
  36.                         pos = shadow.position();
  37.                     }
  39.                     vec.putBase(iov_len, ((DirectBuffer)buf).address() + pos);
  40.                     vec.putLen(iov_len, rem);
  41.                     iov_len++;
  42.                 }
  43.                 i++;
  44.             }
  45.             if (iov_len == 0)
  46.                 return 0L;
  48.             long bytesRead = nd.readv(fd, vec.address, iov_len);
  50.             // Notify the buffers how many bytes were read
  51.             long left = bytesRead;
  52.             for (int j=0; j<iov_len; j++) {
  54.                 ByteBuffer shadow = vec.getShadow(j);
  55.                 if (left > 0) {
  57.                     ByteBuffer buf = vec.getBuffer(j);
  58.                     int rem = vec.getRemaining(j);
  59.                     int n = (left > rem) ? rem : (int)left;
  60.                     //说明buf为直接内存,则修改pos
  61.                     if (shadow == null) {
  63.                         int pos = vec.getPosition(j);
  64.                         buf.position(pos + n);
  65.                     } else {
  67.                         //修改上限
  68.                         shadow.limit(shadow.position() + n);
  69.                         //填充到堆缓冲区
  70.                         buf.put(shadow);
  71.                     }
  72.                     left -= n;
  73.                 }
  74.                 if (shadow != null)
  75.                     //将读取的缓冲放入临时直接缓冲区
  76.                     Util.offerLastTemporaryDirectBuffer(shadow);
  77.                 vec.clearRefs(j);
  78.             }
  80.             completed = true;
  81.             return bytesRead;
  83.         } finally {
  85.             // if an error occurred then clear refs to buffers and return any shadow
  86.             // buffers to cache
  87.             if (!completed) {
  89.                 for (int j=0; j<iov_len; j++) {
  91.                     ByteBuffer shadow = vec.getShadow(j);
  92.                     if (shadow != null)
  93.                         Util.offerLastTemporaryDirectBuffer(shadow);
  94.                     vec.clearRefs(j);
  95.                 }
  96.             }
  97.         }
  98.     }

close

  1. public final void close() throws IOException {
  3.         synchronized (closeLock) {
  5.             if (!open)
  6.                 return;
  7.             //标记关闭
  8.             open = false;
  9.             //调用子类实现
  10.             implCloseChannel();
  11.         }
  12.     }
  14. protected final void implCloseChannel() throws IOException {
  16.         //关闭当前channel
  17.         implCloseSelectableChannel();
  18.         synchronized (keyLock) {
  20.             int count = (keys == null) ? 0 : keys.length;
  21.             for (int i = 0; i < count; i++) {
  23.                 SelectionKey k = keys[i];
  24.                 if (k != null)
  25.                     //注册的channel全部取消
  26.                     k.cancel();
  27.             }
  28.         }
  29.     }
  31. protected void implCloseSelectableChannel() throws IOException {
  33.         synchronized (stateLock) {
  35.             //标记读写关闭,因为基于TCP所以支持双向读写操作
  36.             isInputOpen = false;
  37.             isOutputOpen = false;
  39.             if (state != ST_KILLED)
  40.                 //windows不做处理
  41.                 //linux和Solaris需要，关闭前将fd复制到另一个待关闭fd中，以防止被fd回收
  42.                 nd.preClose(fd);
  44.             if (readerThread != 0)
  45.                 //发送信号给读线程，将其从阻塞I/O中释放，避免一直被阻塞。
  46.                 NativeThread.signal(readerThread);
  48.             if (writerThread != 0)
  49.                 //发送信号给写线程，将其从阻塞I/O中释放，避免一直被阻塞。
  50.                 NativeThread.signal(writerThread);
  52.             //若还有注册的channel，则不处理，等待key全部注销后再kill
  53.             //若没有的话可以直接kill当前channel
  54.             if (!isRegistered())
  55.                 kill();
  56.         }
  57.     }
  59.     public void kill() throws IOException {
  61.         synchronized (stateLock) {
  63.             if (state == ST_KILLED)
  64.                 return;
  65.             if (state == ST_UNINITIALIZED) {
  67.                 state = ST_KILLED;
  68.                 return;
  69.             }
  70.             assert !isOpen() && !isRegistered();
  72.             // Postpone the kill if there is a waiting reader
  73.             // or writer thread. See the comments in read() for
  74.             // more detailed explanation.
  75.             //若仍有线程还没释放，则等线程I/O执行完后再kill
  76.             if (readerThread == 0 && writerThread == 0) {
  78.                 nd.close(fd);
  79.                 state = ST_KILLED;
  80.             } else {
  82.                 state = ST_KILLPENDING;
  83.             }
  84.         }
  85.     }

关闭读写流

  1. //关闭输入流
  2.     @Override
  3.     public SocketChannel shutdownInput() throws IOException {
  5.         synchronized (stateLock) {
  7.             if (!isOpen())
  8.                 throw new ClosedChannelException();
  9.             if (!isConnected())
  10.                 throw new NotYetConnectedException();
  11.             if (isInputOpen) {
  13.                 Net.shutdown(fd, Net.SHUT_RD);
  14.                 if (readerThread != 0)
  15.                     NativeThread.signal(readerThread);
  16.                 isInputOpen = false;
  17.             }
  18.             return this;
  19.         }
  20.     }
  22.     //关闭输出流
  23.     @Override
  24.     public SocketChannel shutdownOutput() throws IOException {
  26.         synchronized (stateLock) {
  28.             if (!isOpen())
  29.                 throw new ClosedChannelException();
  30.             if (!isConnected())
  31.                 throw new NotYetConnectedException();
  32.             if (isOutputOpen) {
  34.                 Net.shutdown(fd, Net.SHUT_WR);
  35.                 if (writerThread != 0)
  36.                     NativeThread.signal(writerThread);
  37.                 isOutputOpen = false;
  38.             }
  39.             return this;
  40.         }
  41.     }

DatagramChannel

open

socket

bind

receive

send

发表评论

表情：
评论列表 （有 0 条评论，81人围观）

还没有评论，来说两句吧...

相关阅读