Finology 大数据金融

通过大数据以量化金融

0%

NIO的通道

发表于 更新于 分类于 ， 阅读次数： 评论数：

通道(Channel)，用于源节点和目标节点的连接。在Java NIO中负责数据的传输。Channel本身不存储数据，所以传输时还需要缓冲区配合。

更多缓冲区的内容，请参考 NIO的缓冲区

利用通道完成文件复制（非直接缓冲区）

可以拷贝超过2G的文件。关于大文件的拷贝，后面再写文章。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
try (FileInputStream fis = new FileInputStream("/infile");
     FileOutputStream fos = new FileOutputStream("/outfile");

     // 获取通道
     FileChannel inChannel = fis.getChannel();
     FileChannel outChannel = fos.getChannel()
) {
    // 分配指定大小的缓冲区
    ByteBuffer buffer = ByteBuffer.allocate(10240);

    // 将通道中的数据存入缓冲区
    while (inChannel.read(buffer) != -1) {
        buffer.flip();
        outChannel.write(buffer);
        buffer.clear();
    }
} catch (IOException e) {
    e.printStackTrace();
}

使用直接缓冲区完成文件复制

文件大小不能超过2G。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
try (FileChannel inChannel = FileChannel.open(Paths.get("/infile"),
        StandardOpenOption.READ);
     FileChannel outChannel = FileChannel.open(Paths.get("/outfile"),
             StandardOpenOption.WRITE, StandardOpenOption.READ, StandardOpenOption.CREATE)) {

    // 内存映射文件
    MappedByteBuffer inMappedByteBuffer = inChannel.map(FileChannel.MapMode.READ_ONLY, 0, inChannel.size());
    MappedByteBuffer outMappedByteBuffer = outChannel.map(FileChannel.MapMode.READ_WRITE, 0, inChannel.size());

    // 直接对缓冲区进行数据的读写操作
    byte[] bytes = new byte[inMappedByteBuffer.limit()];
    inMappedByteBuffer.get(bytes);
    outMappedByteBuffer.put(bytes);
} catch (IOException e) {
    e.printStackTrace();
}

通道之间的数据传输(直接缓冲区)

文件大小不能超过2G。

1
2
3
4
5
6
7
8
9
10
11
12
// transferFrom()
// transferTo()
try (FileChannel inChannel = FileChannel.open(Paths.get("/Users/simon/Documents/largefile.tar"),
        StandardOpenOption.READ);
     FileChannel outChannel = FileChannel.open(Paths.get("/Users/simon/Documents/largefile.tar.copy"),
             StandardOpenOption.WRITE, StandardOpenOption.READ, StandardOpenOption.CREATE)) {

    inChannel.transferTo(0, inChannel.size(), outChannel);

} catch (IOException e) {
    e.printStackTrace();
}

分散和聚集

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
ByteBuffer[] bufs = new ByteBuffer[0];
try (RandomAccessFile raf1 = new RandomAccessFile("/infile", "rw");

     // 1. 获取通道
     FileChannel raf1Channel = raf1.getChannel()) {


    // 2. 分配指定大小的缓冲区数组
    ByteBuffer buf1 = ByteBuffer.allocate(100);
    ByteBuffer buf2 = ByteBuffer.allocate(1024);
    bufs = new ByteBuffer[]{buf1, buf2};

    // 3. 分散读取
    raf1Channel.read(bufs);

    for (ByteBuffer buf : bufs) {
        buf.flip();
    }
} catch (IOException e) {
    e.printStackTrace();
}

System.out.println(new String(bufs[0].array(), 0, bufs[0].limit()));
System.out.println("---");
System.out.println(new String(bufs[1].array(), 0, bufs[1].limit()));

// 4. 聚集写入
try (RandomAccessFile raf2 = new RandomAccessFile("/outfile", "rw");
     FileChannel raf2Channel = raf2.getChannel()) {

    raf2Channel.write(bufs);
} catch (IOException e) {
    e.printStackTrace();
}
相关文章