Java 高并发八：NIO和AIO详解_java

IO感觉上和多线程并没有多大关系，但是NIO改变了线程在应用层面使用的方式，也解决了一些实际的困难。而AIO是异步IO和前面的系列也有点关系。在此，为了学习和记录，也写一篇文章来介绍NIO和AIO。

1. 什么是NIO

NIO是New I/O的简称，与旧式的基于流的I/O方法相对，从名字看，它表示新的一套Java I/O标 准。它是在Java 1.4中被纳入到JDK中的，并具有以下特性：

  1. NIO是基于块（Block）的，它以块为基本单位处理数据 （硬盘上存储的单位也是按Block来存储，这样性能上比基于流的方式要好一些）
  2. 为所有的原始类型提供（Buffer）缓存支持
  3. 增加通道（Channel）对象，作为新的原始 I/O 抽象
  4. 支持锁（我们在平时使用时经常能看到会出现一些.lock的文件，这说明有线程正在使用这把锁，当线程释放锁时，会把这个文件删除掉，这样其他线程才能继续拿到这把锁）和内存映射文件的文件访问接口
  5. 提供了基于Selector的异步网络I/O

所有的从通道中的读写操作，都要经过Buffer，而通道就是io的抽象，通道的另一端就是操纵的文件。

2. Buffer

Java中Buffer的实现。基本的数据类型都有它对应的Buffer

Buffer的简单使用例子：

package test;

import java.io.File;
import java.io.FileInputStream;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;

public class Test {
 public static void main(String[] args) throws Exception {
  FileInputStream fin = new FileInputStream(new File(
    "d:\\temp_buffer.tmp"));
  FileChannel fc = fin.getChannel();
  ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
  fc.read(byteBuffer);
  fc.close();
  byteBuffer.flip();//读写转换
 }
}

总结下使用的步骤是：

1. 得到Channel

2. 申请Buffer

3. 建立Channel和Buffer的读/写关系

4. 关闭

下面的例子是使用NIO来复制文件：

public static void nioCopyFile(String resource, String destination)
   throws IOException {
  FileInputStream fis = new FileInputStream(resource);
  FileOutputStream fos = new FileOutputStream(destination);
  FileChannel readChannel = fis.getChannel(); // 读文件通道
  FileChannel writeChannel = fos.getChannel(); // 写文件通道
  ByteBuffer buffer = ByteBuffer.allocate(1024); // 读入数据缓存
  while (true) {
   buffer.clear();
   int len = readChannel.read(buffer); // 读入数据
   if (len == -1) {
    break; // 读取完毕
   }
   buffer.flip();
   writeChannel.write(buffer); // 写入文件
  }
  readChannel.close();
  writeChannel.close();
 }

Buffer中有3个重要的参数：位置（position）、容量（capactiy）和上限（limit）

这里要区别下容量和上限，比如一个Buffer有10KB，那么10KB就是容量，我将5KB的文件读到Buffer中，那么上限就是5KB。

下面举个例子来理解下这3个重要的参数：

public static void main(String[] args) throws Exception {
  ByteBuffer b = ByteBuffer.allocate(15); // 15个字节大小的缓冲区
  System.out.println("limit=" + b.limit() + " capacity=" + b.capacity()
    + " position=" + b.position());
  for (int i = 0; i < 10; i++) {
   // 存入10个字节数据
   b.put((byte) i);
  }
  System.out.println("limit=" + b.limit() + " capacity=" + b.capacity()
    + " position=" + b.position());
  b.flip(); // 重置position
  System.out.println("limit=" + b.limit() + " capacity=" + b.capacity()
    + " position=" + b.position());
  for (int i = 0; i < 5; i++) {
   System.out.print(b.get());
  }
  System.out.println();
  System.out.println("limit=" + b.limit() + " capacity=" + b.capacity()
    + " position=" + b.position());
  b.flip();
  System.out.println("limit=" + b.limit() + " capacity=" + b.capacity()
    + " position=" + b.position());

 }

整个过程如图：

此时position从0到10，capactiy和limit不变。

该操作会重置position，通常，将buffer从写模式转换为读 模式时需要执行此方法 flip()操作不仅重置了当前的position为0，还将limit设置到当前position的位置 。

limit的意义在于，来确定哪些数据是有意义的，换句话说，从position到limit之间的数据才是有意义的数据，因为是上次操作的数据。所以flip操作往往是读写转换的意思。

意义同上。

而Buffer中大多数的方法都是去改变这3个参数来达到某些功能的：

public final Buffer rewind()

将position置零，并清除标志位（mark）

public final Buffer clear()

将position置零，同时将limit设置为capacity的大小，并清除了标志mark

public final Buffer flip()

先将limit设置到position所在位置，然后将position置零，并清除标志位mark，通常在读写转换时使用

文件映射到内存

public static void main(String[] args) throws Exception {
  RandomAccessFile raf = new RandomAccessFile("C:\\mapfile.txt", "rw");
  FileChannel fc = raf.getChannel();
  // 将文件映射到内存中
  MappedByteBuffer mbb = fc.map(FileChannel.MapMode.READ_WRITE, 0,
    raf.length());
  while (mbb.hasRemaining()) {
   System.out.print((char) mbb.get());
  }
  mbb.put(0, (byte) 98); // 修改文件
  raf.close();
 }

对MappedByteBuffer的修改就相当于修改文件本身，这样操作的速度是很快的。

3. Channel

多线程网络服务器的一般结构：

简单的多线程服务器：

public static void main(String[] args) throws Exception {
  ServerSocket echoServer = null;
  Socket clientSocket = null;
  try {
   echoServer = new ServerSocket(8000);
  } catch (IOException e) {
   System.out.println(e);
  }
  while (true) {
   try {
    clientSocket = echoServer.accept();
    System.out.println(clientSocket.getRemoteSocketAddress()
      + " connect!");
    tp.execute(new HandleMsg(clientSocket));
   } catch (IOException e) {
    System.out.println(e);
   }
  }
 }

功能就是服务器端读到什么数据，就向客户端回写什么数据。

这里的tp是一个线程池，HandleMsg是处理消息的类。

static class HandleMsg implements Runnable{
   省略部分信息
   public void run(){
    try {
     is = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()));
     os = new PrintWriter(clientSocket.getOutputStream(), true);
     // 从InputStream当中读取客户端所发送的数据
     String inputLine = null;
     long b=System. currentTimeMillis ();
     while ((inputLine = is.readLine()) != null)
     {
      os.println(inputLine);
     }
     long e=System. currentTimeMillis ();
     System. out.println ("spend:"+(e - b)+" ms ");
   } catch (IOException e) {
    e.printStackTrace();
   }finally
   {
    关闭资源
   }
  }
  }

客户端：

public static void main(String[] args) throws Exception {
  Socket client = null;
  PrintWriter writer = null;
  BufferedReader reader = null;
  try {
   client = new Socket();
   client.connect(new InetSocketAddress("localhost", 8000));
   writer = new PrintWriter(client.getOutputStream(), true);
   writer.println("Hello!");
   writer.flush();
   reader = new BufferedReader(new InputStreamReader(
     client.getInputStream()));
   System.out.println("from server: " + reader.readLine());
  } catch (Exception e) {
  } finally {
   // 省略资源关闭
  }
 }

以上的网络编程是很基本的，使用这种方式，会有一些问题：

为每一个客户端使用一个线程，如果客户端出现延时等异常，线程可能会被占用很长时间。因为数据的准备和读取都在这个线程中。此时，如果客户端数量众多，可能会消耗大量的系统资源。

解决方案:

使用非阻塞的NIO （读取数据不等待，数据准备好了再工作）

为了体现NIO使用的高效。

这里先模拟一个低效的客户端来模拟因网络而延时的情况：

private static ExecutorService tp= Executors.newCachedThreadPool();
  private static final int sleep_time=1000*1000*1000;
  public static class EchoClient implements Runnable{
   public void run(){
    try {
     client = new Socket();
     client.connect(new InetSocketAddress("localhost", 8000));
     writer = new PrintWriter(client.getOutputStream(), true);
     writer.print("H");
     LockSupport.parkNanos(sleep_time);
     writer.print("e");
     LockSupport.parkNanos(sleep_time);
     writer.print("l");
     LockSupport.parkNanos(sleep_time);
     writer.print("l");
     LockSupport.parkNanos(sleep_time);
     writer.print("o");
     LockSupport.parkNanos(sleep_time);
     writer.print("!");
     LockSupport.parkNanos(sleep_time);
     writer.println();
     writer.flush();
    }catch(Exception e)
    {
    }
   }
  }

服务器端输出：

spend:6000ms
spend:6000ms
spend:6000ms
spend:6001ms
spend:6002ms
spend:6002ms
spend:6002ms
spend:6002ms
spend:6003ms
spend:6003ms

因为

while ((inputLine = is.readLine()) != null)

是阻塞的，所以时间都花在等待中。

如果用NIO来处理这个问题会怎么做呢？

NIO有一个很大的特点就是：把数据准备好了再通知我

而Channel有点类似于流，一个Channel可以和文件或者网络Socket对应 。

selector是一个选择器，它可以选择某一个Channel，然后做些事情。

一个线程可以对应一个selector，而一个selector可以轮询多个Channel，而每个Channel对应了一个Socket。

与上面一个线程对应一个Socket相比，使用NIO后，一个线程可以轮询多个Socket。

当selector调用select()时，会查看是否有客户端准备好了数据。当没有数据被准备好时，select()会阻塞。平时都说NIO是非阻塞的，但是如果没有数据被准备好还是会有阻塞现象。

当有数据被准备好时，调用完select()后，会返回一个SelectionKey，SelectionKey表示在某个selector上的某个Channel的数据已经被准备好了。

只有在数据准备好时，这个Channel才会被选择。

这样NIO实现了一个线程来监控多个客户端。

而刚刚模拟的网络延迟的客户端将不会影响NIO下的线程，因为某个Socket网络延迟时，数据还未被准备好，selector是不会选择它的，而会选择其他准备好的客户端。

selectNow()与select()的区别在于，selectNow()是不阻塞的，当没有客户端准备好数据时，selectNow()不会阻塞，将返回0，有客户端准备好数据时，selectNow()返回准备好的客户端的个数。

主要代码：

package test;

import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.channels.spi.AbstractSelector;
import java.nio.channels.spi.SelectorProvider;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class MultiThreadNIOEchoServer {
 public static Map<Socket, Long> geym_time_stat = new HashMap<Socket, Long>();

 class EchoClient {
  private LinkedList<ByteBuffer> outq;

  EchoClient() {
   outq = new LinkedList<ByteBuffer>();
  }

  public LinkedList<ByteBuffer> getOutputQueue() {
   return outq;
  }

  public void enqueue(ByteBuffer bb) {
   outq.addFirst(bb);
  }
 }

 class HandleMsg implements Runnable {
  SelectionKey sk;
  ByteBuffer bb;

  public HandleMsg(SelectionKey sk, ByteBuffer bb) {
   super();
   this.sk = sk;
   this.bb = bb;
  }

  @Override
  public void run() {
   // TODO Auto-generated method stub
   EchoClient echoClient = (EchoClient) sk.attachment();
   echoClient.enqueue(bb);
   sk.interestOps(SelectionKey.OP_READ | SelectionKey.OP_WRITE);
   selector.wakeup();
  }

 }

 private Selector selector;
 private ExecutorService tp = Executors.newCachedThreadPool();

 private void startServer() throws Exception {
  selector = SelectorProvider.provider().openSelector();
  ServerSocketChannel ssc = ServerSocketChannel.open();
  ssc.configureBlocking(false);
  InetSocketAddress isa = new InetSocketAddress(8000);
  ssc.socket().bind(isa);
  // 注册感兴趣的事件，此处对accpet事件感兴趣
  SelectionKey acceptKey = ssc.register(selector, SelectionKey.OP_ACCEPT);
  for (;;) {
   selector.select();
   Set readyKeys = selector.selectedKeys();
   Iterator i = readyKeys.iterator();
   long e = 0;
   while (i.hasNext()) {
    SelectionKey sk = (SelectionKey) i.next();
    i.remove();
    if (sk.isAcceptable()) {
     doAccept(sk);
    } else if (sk.isValid() && sk.isReadable()) {
     if (!geym_time_stat.containsKey(((SocketChannel) sk
       .channel()).socket())) {
      geym_time_stat.put(
        ((SocketChannel) sk.channel()).socket(),
        System.currentTimeMillis());
     }
     doRead(sk);
    } else if (sk.isValid() && sk.isWritable()) {
     doWrite(sk);
     e = System.currentTimeMillis();
     long b = geym_time_stat.remove(((SocketChannel) sk
       .channel()).socket());
     System.out.println("spend:" + (e - b) + "ms");
    }
   }
  }
 }

 private void doWrite(SelectionKey sk) {
  // TODO Auto-generated method stub
  SocketChannel channel = (SocketChannel) sk.channel();
  EchoClient echoClient = (EchoClient) sk.attachment();
  LinkedList<ByteBuffer> outq = echoClient.getOutputQueue();
  ByteBuffer bb = outq.getLast();
  try {
   int len = channel.write(bb);
   if (len == -1) {
    disconnect(sk);
    return;
   }
   if (bb.remaining() == 0) {
    outq.removeLast();
   }
  } catch (Exception e) {
   // TODO: handle exception
   disconnect(sk);
  }
  if (outq.size() == 0) {
   sk.interestOps(SelectionKey.OP_READ);
  }
 }

 private void doRead(SelectionKey sk) {
  // TODO Auto-generated method stub
  SocketChannel channel = (SocketChannel) sk.channel();
  ByteBuffer bb = ByteBuffer.allocate(8192);
  int len;
  try {
   len = channel.read(bb);
   if (len < 0) {
    disconnect(sk);
    return;
   }
  } catch (Exception e) {
   // TODO: handle exception
   disconnect(sk);
   return;
  }
  bb.flip();
  tp.execute(new HandleMsg(sk, bb));
 }

 private void disconnect(SelectionKey sk) {
  // TODO Auto-generated method stub
  //省略略干关闭操作
 }

 private void doAccept(SelectionKey sk) {
  // TODO Auto-generated method stub
  ServerSocketChannel server = (ServerSocketChannel) sk.channel();
  SocketChannel clientChannel;
  try {
   clientChannel = server.accept();
   clientChannel.configureBlocking(false);
   SelectionKey clientKey = clientChannel.register(selector,
     SelectionKey.OP_READ);
   EchoClient echoClinet = new EchoClient();
   clientKey.attach(echoClinet);
   InetAddress clientAddress = clientChannel.socket().getInetAddress();
   System.out.println("Accepted connection from "
     + clientAddress.getHostAddress());
  } catch (Exception e) {
   // TODO: handle exception
  }
 }

 public static void main(String[] args) {
  // TODO Auto-generated method stub
  MultiThreadNIOEchoServer echoServer = new MultiThreadNIOEchoServer();
  try {
   echoServer.startServer();
  } catch (Exception e) {
   // TODO: handle exception
  }

 }

}

代码仅作参考，主要的特点是，对不同事件的感兴趣来做不同的事。

当用之前模拟的那个延迟的客户端时，这次的时间消耗就在2ms到11ms之间了。性能提升是很明显的。

总结：

1. NIO会将数据准备好后，再交由应用进行处理，数据的读取/写入过程依然在应用线程中完成，只是将等待的时间剥离到单独的线程中去。

2. 节省数据准备时间（因为Selector可以复用）

5. AIO

AIO的特点：

1. 读完了再通知我

2. 不会加快IO，只是在读完后进行通知

3. 使用回调函数，进行业务处理

AIO的相关代码：

AsynchronousServerSocketChannel

server = AsynchronousServerSocketChannel.open().bind( new InetSocketAddress (PORT));
使用server上的accept方法

public abstract <A> void accept(A attachment,CompletionHandler<AsynchronousSocketChannel,? super A> handler);
CompletionHandler为回调接口，当有客户端accept之后，就做handler中的事情。

示例代码：

server.accept(null,
    new CompletionHandler<AsynchronousSocketChannel, Object>() {
     final ByteBuffer buffer = ByteBuffer.allocate(1024);

     public void completed(AsynchronousSocketChannel result,
       Object attachment) {
      System.out.println(Thread.currentThread().getName());
      Future<Integer> writeResult = null;
      try {
       buffer.clear();
       result.read(buffer).get(100, TimeUnit.SECONDS);
       buffer.flip();
       writeResult = result.write(buffer);
      } catch (InterruptedException | ExecutionException e) {
       e.printStackTrace();
      } catch (TimeoutException e) {
       e.printStackTrace();
      } finally {
       try {
        server.accept(null, this);
        writeResult.get();
        result.close();
       } catch (Exception e) {
        System.out.println(e.toString());
       }
      }
     }

     @Override
     public void failed(Throwable exc, Object attachment) {
      System.out.println("failed: " + exc);
     }
    });

这里使用了Future来实现即时返回，关于Future请参考上一篇

在理解了NIO的基础上，看AIO，区别在于AIO是等读写过程完成后再去调用回调函数。

NIO是同步非阻塞的

AIO是异步非阻塞的

由于NIO的读写过程依然在应用线程里完成，所以对于那些读写过程时间长的，NIO就不太适合。

而AIO的读写过程完成后才被通知，所以AIO能够胜任那些重量级，读写过程长的任务。

以上是小编为您精心准备的的内容，在的博客、问答、公众号、人物、课程等栏目也有的相关内容，欢迎继续使用右上角搜索按钮进行搜索java
高并发
java官网、java教程、java下载、java7、java jdk，以便于您获取更多的相关知识。

时间： 2024-08-22 14:12:40

Java 高并发八：NIO和AIO详解_java的相关文章

Java并发编程总结——慎用CAS详解_java

一.CAS和synchronized适用场景 1.对于资源竞争较少的情况,使用synchronized同步锁进行线程阻塞和唤醒切换以及用户态内核态间的切换操作额外浪费消耗cpu资源:而CAS基于硬件实现,不需要进入内核,不需要切换线程,操作自旋几率较少,因此可以获得更高的性能. 2.对于资源竞争严重的情况,CAS自旋的概率会比较大,从而浪费更多的CPU资源,效率低于synchronized.以java.util.concurrent.atomic包中AtomicInteger类为例,其getAn

Java 高并发四：无锁详细介绍_java

在[高并发Java 一] 前言中已经提到了无锁的概念,由于在jdk源码中有大量的无锁应用,所以在这里介绍下无锁. 1 无锁类的原理详解 1.1 CAS CAS算法的过程是这样:它包含3个参数CAS(V,E,N).V表示要更新的变量,E表示预期值,N表示新值.仅当V 值等于E值时,才会将V的值设为N,如果V值和E值不同,则说明已经有其他线程做了更新,则当前线程什么 都不做.最后,CAS返回当前V的真实值.CAS操作是抱着乐观的态度进行的,它总是认为自己可以成功完成 操作.当多个线程同时使用CAS操

Java 高并发二：多线程基础详细介绍_java

本系列基于炼数成金课程,为了更好的学习,做了系列的记录. 本文主要介绍 1.什么是线程 2.线程的基本操作 3.守护线程 4.线程优先级 5.基本的线程同步操作 1. 什么是线程 线程是进程内的执行单元 某个进程当中都有若干个线程. 线程是进程内的执行单元. 使用线程的原因是,进程的切换是非常重量级的操作,非常消耗资源.如果使用多进程,那么并发数相对来说不会很高.而线程是更细小的调度单元,更加轻量级,所以线程会较为广泛的用于并发设计. 在Java当中线程的概念和操作系统级别线程的概念是类似的.事

Java的JSON处理器fastjson使用方法详解_java

fastjson 是一个性能很好的 Java 语言实现的 JSON 解析器和生成器,来自阿里巴巴的工程师开发. 主要特点: • 快速FAST (比其它任何基于Java的解析器和生成器更快,包括jackson) • 强大(支持普通JDK类包括任意Java Bean Class.Collection.Map.Date或enum) • 零依赖(没有依赖其它任何类库除了JDK) 示例代码: import com.alibaba.fastjson.JSON; Group group = new Group

java中哈希表及其应用详解_java

哈希表也称为散列表,是用来存储群体对象的集合类结构. 什么是哈希表 数组和向量都可以存储对象,但对象的存储位置是随机的,也就是说对象本身与其存储位置之间没有必然的联系.当要查找一个对象时,只能以某种顺序(如顺序查找或二分查找)与各个元素进行比较,当数组或向量中的元素数量很多时,查找的效率会明显的降低. 一种有效的存储方式,是不与其他元素进行比较,一次存取便能得到所需要的记录.这就需要在对象的存储位置和对象的关键属性(设为 k)之间建立一个特定的对应关系(设为 f),使每个对象与一个唯一的存储位置

Java太阳系小游戏分析和源码详解_java

最近看了面向对象的一些知识,然后跟着老师的讲解做了一个太阳系各行星绕太阳转的小游戏,来练习巩固一下最近学的知识: 用到知识点:类的继承.方法的重载与重写.多态.封装等 分析: 1.需要加载图片.画图 2.建一个面板,主页面 3.行星类 效果图: 先看一下源码结构图: 现在逐步分析各个类的功能: 1)工具类-----util包中     --Constant类   封装了游戏中用到的常量     --GameUtil类  封装了游戏的图片加载功能     --MyFrame类  封装了游戏面板的构

JAVA中static方法的用法实例详解_java

本文实例讲述了JAVA中static方法的用法.分享给大家供大家参考,具体如下: static表示"全局"或者"静态"的意思,用来修饰成员变量和成员方法,也可以形成静态static代码块,但是Java语言中没有全局变量的概念. 被static修饰的成员变量和成员方法独立于该类的任何对象.也就是说,它不依赖类特定的实例,被类的所有实例共享.只要这个类被加载,Java虚拟机就能根据类名在运行时数据区或者方法区内找到他们.因此,static对象可以在它的任何对象创建之前访

Java并发编程之阻塞队列详解_java

1.什么是阻塞队列? 队列是一种数据结构,它有两个基本操作:在队列尾部加入一个元素,从队列头部移除一个元素.阻塞队里与普通的队列的区别在于,普通队列不会对当前线程产生阻塞,在面对类似消费者-生产者模型时,就必须额外的实现同步策略以及线程间唤醒策略.使用阻塞队列,就会对当前线程产生阻塞,当队列是空时,从队列中获取元素的操作将会被阻塞,当队列是满时,往队列里添加元素的操作也会被阻塞. 2.主要的阻塞队列及其方法 java.util.concurrent包下提供主要的几种阻塞队列,主要有以下几个: 1

高并发网络编程之epoll详解

select.poll和epoll的区别 在linux没有实现epoll事件驱动机制之前,我们一般选择用select或者poll等IO多路复用的方法来实现并发服务程序.在大数据.高并发.集群等一些名词唱的火热之年代,select和poll的用武之地越来越有限了,风头已经被epoll占尽. select()和poll() IO多路复用模型 select的缺点: 单个进程能够监视的文件描述符的数量存在最大限制,通常是1024,当然可以更改数量,但由于select采用轮询的方式扫描文件描述符,文件描述