Java--NIO之通道

Java NIO
  1. 打开通道
  2. 使用通道
  3. Scatter/Gather

　　通道与缓冲区不同，通道 API 主要由接口指定。不同的操作系统上通道实现（Channel Implementation）会有根本性的差异，所以通道 API 仅仅描述了可以做什么。因此很自然地，通道实现经常使用操作系统的本地代码。通道接口允许您以一种受控且可移植的方式来访问底层的 I/O服务。
　Channel结构关系图

首先，我们来更近距离地看一下基本的 Channel 接口。下面是 Channel 接口的完整源码：

1
2
3
4
5
package java.nio.channels;
public interface Channel{
	public boolean isOpen( );
	public void close() throws IOException;
}

从 Channel 接口引申出的其他接口都是面向字节的子接口，包括 Writable ByteChannel 和ReadableByteChannel。这也正好支持了我们之前所学的：通道只能在字节缓冲区上操作。层次结构表明其他数据类型的通道也可以从 Channel 接口引申而来。这是一种很好的类设计，不过非字节实现是不可能的，因为操作系统都是以字节的形式实现底层 I/O 接口的。

打开通道

通道是访问 I/O 服务的导管，打开一个管道，通常就是创建一个管道。I/O 可以分为广义的两大类别：File I/O 和 Stream I/O。那么相应地也有两种类型的通道：文件（file）通道和套接字（socket）通道。通道可以以多种方式创建。

  1. Socket： 通道有可以直接创建新 socket 通道的工厂方法，分别是SocketChannel、ServerSocketChannel和DatagramChannel。
  2. FileChannel： 对象却只能通过在一个打开的 RandomAccessFile、FileInputStream 或 FileOutputStream对象上调用getChannel()方法来获取，不能直接创建。
  • 打开管道。例如
1
2
3
4
5
6
7
8
9
//Socket（套接字）
SocketChannel sc = SocketChannel.open();
ServerSocketChannel ssc = ServerSocketChannel.open();
DatagramChannel dc = DatagramChannel.open();

//FileChannel
FileChannel raf = new RandomAccessFile("file", "r").getChannel();
FileChannel channel = new FileInputStream("abc.txt").getChannel();
FileChannel channe = new FileOutputStream("abc.txt").getChannel();

使用通道

打开一个管道之后就是管道的字节读写，有 ReadableByteChannel, WritableByteChannel,和InterruptibleChannel。ByteChannel类继承了ReadableByteChannel和WritableByteChannel的属性。然后还有很多很多的Channel了。例如：SocketChannel，FileChannel，ServerSocketChannel等等。
ByteChannel 接口

1
2
3
4
5
6
7
8
9
10
11
public interface ReadableByteChannel extends Channel {
	public int read(ByteBuffer dst) throws IOException;
}

public interface WritableByteChannel extends Channel { 
	public int write(ByteBuffer src) throws IOException;
}

public interface ByteChannel extends 
	ReadableByteChannel, WritableByteChannel {
}
  • 实例代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
package com.xxo.momo.nio;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.Channels;
import java.nio.channels.ReadableByteChannel;
import java.nio.channels.WritableByteChannel;

/**
 *
 * 通道
 * Created by xiaoxiaomo on 2016/4/23.
 */
public class ChannelTest01 {

    public static void main(String[] args) throws IOException {

        //1. 创建通道
        ReadableByteChannel rbyc = Channels.newChannel(System.in);
        WritableByteChannel wbyc = Channels.newChannel(System.out);

        detailChnnel(rbyc, wbyc);
        //detailChnnel2(rbyc, wbyc);//第二种方法

        //4. 关闭
        rbyc.close();
        wbyc.close();
    }

    public static void detailChnnel(
            ReadableByteChannel rbyc, WritableByteChannel wbyc) throws IOException {
        //2. 创建缓冲区
        ByteBuffer buffer = ByteBuffer.allocate(1024);

        //3. 处理数据-输入并打印到控制台
        while ( rbyc.read( buffer ) != -1 ){//读入数据到缓冲区

            //翻转-准备读出缓冲区的数据
            buffer.flip() ;

            wbyc.write( buffer ) ;//从缓冲区写出到控制台

            //压缩-把已经读过的数据抛弃，使用后面的数据覆盖（移动至索引0）
            buffer.compact() ;
        }

        System.out.println("结束程序后--------");
        while ( buffer.hasRemaining() ){//告知在当前位置和限制之间是否有元素。
            wbyc.write( buffer ) ;//
        }
    }

    public static void detailChnnel2(
            ReadableByteChannel rbyc, WritableByteChannel wbyc) throws IOException {

        //2. 创建缓冲区
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        while (rbyc.read(buffer) != -1) {
            buffer.flip();
            while (buffer.hasRemaining()) {
                wbyc.write(buffer);//从缓冲区写出到控制台
            }
            //重设缓冲区以便接收更多的字节
            buffer.compact();
            //buffer.clear();
        }
    }
}
  • SocketChannel和FileChannel都是实现了以上三个接口的。那么，也就是说，他们都有读写的方法，但是对于FileChannel来说，并非如此，因为有的文件可能是只读的，那么，如果调用write方法的，就会抛出NonWritableChannelException异常。

Scatter/Gather

有时候可能为了提高I/O性能，要组合多个缓冲区，或者从多个缓冲区读数据，就可以使用Scatter（分散）和Gather（聚合）

  1. Scatter：从一个Channel读取的信息分散到N个缓冲区中Buufer。
  2. Gather：将N个Buffer里面内容按照顺序发送到一个Channel。

Scatter/Gather

  • ReadableByteChannel、WritableByteChannel接口提供了通道的读写功能
  • ScatteringByteChannel、GatheringByteChannel接口提供了分散和聚合
1
2
3
4
5
6
7
8
9
10

public interface ScatteringByteChannel extends ReadableByteChannel{
  public long read (ByteBuffer [] dsts) throws IOException;
  public long read (ByteBuffer [] dsts, int offset, int length) throws IOException; 
}

public interface GatheringByteChannel extends WritableByteChannel{ 
  public long write(ByteBuffer[] srcs) throws IOException; 
  public long write(ByteBuffer[] srcs, int offset, int length) throws IOException; 
}
  • 下面以FileChannel为例（FileChannel implements GatheringByteChannel, ScatteringByteChannel）
  • 实例Scatter
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
package com.xxo.momo.nio;

import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.charset.Charset;

/**
 * scatter 分散
 * Created by xiaoxiaomo on 2016/4/23.
 */
public class ScatterDemo {

    private static Charset charset = Charset.forName("GBK");

    public static void main(String[] args) throws IOException {
        final String fileName = "D:/test.log";
        //写入数据
        writeData(fileName, "abcde999efghigk12345678");

        /**----------Scatter------------*/
        //read(java.nio.ByteBuffer[])
        scatter(fileName);

        //read(java.nio.ByteBuffer[], int, int)
        scatter2(fileName);
    }

    /**
     * Scatter 分散
     * @param fileName
     * @throws IOException
     */
    private static void scatter(final String fileName) throws IOException {
        
        //1. 获取文件通道
        RandomAccessFile accessFile = new RandomAccessFile(fileName, "r");
        FileChannel channel = accessFile.getChannel();
        
        //2. 创建两个缓冲区
        ByteBuffer head = ByteBuffer.allocate(5);
        ByteBuffer body = ByteBuffer.allocate(1024);

        //3. 缓冲区数组
        ByteBuffer[] buffers = {head, body};
        // channel读取的信息分散到head和body缓冲区中
        // head 前5个字节 |  body 剩下的
        System.out.println("scatter====共读到多少字节:" + channel.read(buffers));

        //head缓冲区中的数据:abcde
        head.flip();
        System.out.println("head缓冲区中的数据:" + charset.decode(head));

        //body缓冲区中的数据:999efghigk12345678
        body.flip();
        System.out.println("body缓冲区中的数据:" + charset.decode(body));

        //4. 关闭通道
        accessFile.close();
        channel.close();
    }

    /**
     * scatter2
     * @param fileName
     * @throws IOException
     */
    private static void scatter2(final String fileName) throws IOException {

        //1. 获取文件通道
        RandomAccessFile accessFile = new RandomAccessFile(fileName, "r");
        FileChannel channel = accessFile.getChannel();

        //2. 创建一个四个缓冲区
        ByteBuffer head = ByteBuffer.allocate(5);
        ByteBuffer body1 = ByteBuffer.allocate(3);
        ByteBuffer body2 = ByteBuffer.allocate(5);
        ByteBuffer body3 = ByteBuffer.allocate(2);

        ByteBuffer[] buffers = new ByteBuffer[]{ head, body1, body2, body3};
        //0从那个缓冲区开始被使用，使用3个缓冲区即head,body1,body2
        System.out.println("scatter2====共读到多少字节:" + channel.read(buffers, 0, 3));

        //head缓冲区中的数据:abcde
        head.flip();
        System.out.println("head缓冲区中的数据:" + charset.decode(head));

        //body1缓冲区中的数据:999
        body1.flip();
        System.out.println("body1缓冲区中的数据:" + charset.decode(body1));

        //body2缓冲区中的数据:efghi
        body2.flip();
        System.out.println("body2缓冲区中的数据:" + charset.decode(body2));

        //body3，没有数据
        body3.flip();
        System.out.println("body3缓冲区中的数据:" + charset.decode(body3));

        //4. 关闭流
        accessFile.close();
        channel.close();
    }

    /**
     * 写入数据
     * writeData
     * @param fileName
     * @param data
     * @throws IOException
     */
    private static void writeData(final String fileName, String data) throws IOException {
        RandomAccessFile accessFile = new RandomAccessFile(fileName, "rw");
        accessFile.writeBytes(data);
        accessFile.close();
    }
}

//scatter====共读到多少字节:23
//head缓冲区中的数据:abcde
//body缓冲区中的数据:999efghigk12345678
//scatter2====共读到多少字节:13
//head缓冲区中的数据:abcde
//body1缓冲区中的数据:999
//body2缓冲区中的数据:efghi
//body3缓冲区中的数据:
  • 实例Gather
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
package com.xxo.momo.nio;

import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.charset.Charset;

/**
 * Gather 聚合
 * Created by xiaoxiaomo on 2016/4/23.
 */
public class GatherDemo {

    public static void main(String[] args) throws IOException {
        final String fileName = "D:/test2.log";
        /**----------Gather------------*/
        //FileChannel#write(java.nio.ByteBuffer[])
        gather(fileName);

        //FileChannel#write(java.nio.ByteBuffer[], int, int)
        gather2(fileName);
    }

    /**
     * gather
     * @param fileName
     * @throws IOException
     */
    private static void gather(String fileName) throws IOException {
        //1. 获取文件通道
        RandomAccessFile accessFile = new RandomAccessFile(fileName, "rw");
        FileChannel channel = accessFile.getChannel();

        //2. 创建两个缓冲区head,body 并写入数据
        ByteBuffer head = ByteBuffer.allocate(5);
        head.put("abcde".getBytes());

        ByteBuffer body = ByteBuffer.allocate(1024);
        body.put("999efghigk12345678".getBytes());

        ByteBuffer[] allBuffers = { head, body };

        head.flip();
        body.flip();

        //将按allBuffers顺序  写入abcde999efghigk12345678
        System.out.println("gather====共写入多少字节:" + channel.write(allBuffers));

        //3. 关闭
        accessFile.close();
        channel.close();
    }

    /**
     * gather2
     * @param fileName
     * @throws IOException
     */
    private static void gather2(String fileName) throws IOException {

        //1. 获取文件通道
        RandomAccessFile accessFile = new RandomAccessFile(fileName, "rw");
        FileChannel channel = accessFile.getChannel();

        //2. 创建四个缓冲区
        ByteBuffer head = ByteBuffer.allocate(5);
        ByteBuffer body1 = ByteBuffer.allocate(3);
        ByteBuffer body2 = ByteBuffer.allocate(5);
        ByteBuffer body3 = ByteBuffer.allocate(20);

        head.put("abcde".getBytes());
        body1.put("999".getBytes());
        body2.put("efghi".getBytes());
        body3.put("gk12345678".getBytes());

        ByteBuffer[] allBuffers = new ByteBuffer[]{
            head, body1, body2, body3};

        head.flip();
        body1.flip();
        body2.flip();
        body3.flip();

        //将按allBuffers数组顺序使用两个缓冲区
        //0从哪开始
        //2使用几个
        //当前使用head  body1
        //最终写入abcdefg
        long n = channel.write(allBuffers, 0, 2);

        //应该返回8个字节
        System.out.println("gather2====共写入多少字节:" + n);

        accessFile.close();
        channel.close();
    }
}
//运行结果
//gather====共写入多少字节:23
//gather2====共写入多少字节:8

当前网速较慢或者你使用的浏览器不支持博客特定功能，请尝试刷新或换用Chrome、Firefox等现代浏览器