java - FileChannel 零拷贝 transferTo 拷贝字节到 SocketChannel 失败

Question

在 Java 中使用零拷贝将大文件从文件传输到套接字时，我看到了一些奇怪的行为。我的环境:

• Windows 7 64 位 JDK 1.6.0_45 和 1.7.0_79。
• Centos 6.6 64 位 JDK 1.6.0_35

程序的作用:客户端将输入文件复制到套接字中，服务器使用零复制方法将套接字复制到输出文件:transferFrom 和 transferTo。如果文件大小相对较大，则并非所有字节都到达服务器，Windows 为 100Mb+，Centos 为 2GB+。客户端和服务器驻留在同一台机器上，本地主机地址用于传输数据。

行为因操作系统而异。在 Windows 上，客户端成功完成 transferTo 方法。传输的字节数等于输入文件大小。

long bytesTransferred = fileChannel.transferTo(0, inputFile.length(), socketChannel);

另一方面，服务器报告收到的字节数较少。

long transferFromByteCount = fileChannel.transferFrom(socketChannel, 0, inputFile.length());

在 Linux 上，即使输入文件大小为 4Gb，客户端上传输的字节数也是 2Gb。两种配置都有足够的空间。

在 Windows 上，我能够使用以下解决方法之一传输 130Mb 文件:1) 增加服务器上的接收缓冲区大小和 2) 在客户端中添加线程 hibernate 方法。这使我认为当所有字节都发送到套接字发送缓冲区而不是服务器时，客户端上的 transferTo 方法完成。无法保证这些字节是否到达服务器，这会给我的用例带来问题。

在 Linux 上，我能够通过单个 transferTo 调用传输的最大文件大小为 2Gb，但是至少客户端报告发送到服务器的字节数是正确的。

我的问题:客户端确保跨平台将文件传送到服务器的最佳方式是什么？在 Windows 上使用什么机制来模拟 sendfile()？

代码如下:

客户端 - ZeroCopyClient.java:

import org.apache.commons.io.FileUtils;

import java.io.*;
import java.net.*;
import java.nio.channels.*;

public class ZeroCopyClient {

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

        final File inputFile = new File(args[0]);

        FileInputStream fileInputStream = new FileInputStream(inputFile);
        FileChannel fileChannel = fileInputStream.getChannel();
        SocketAddress socketAddress = new InetSocketAddress("localhost", 8083);
        SocketChannel socketChannel = SocketChannel.open();
        socketChannel.connect(socketAddress);

        System.out.println("sending " + inputFile.length() + " bytes to " + socketChannel);

        long startTime = System.currentTimeMillis();
        long totalBytesTransferred = 0;
        while (totalBytesTransferred < inputFile.length()) {
            long st = System.currentTimeMillis();
            long bytesTransferred = fileChannel.transferTo(totalBytesTransferred, inputFile.length()-totalBytesTransferred, socketChannel);
            totalBytesTransferred += bytesTransferred;
            long et = System.currentTimeMillis();
            System.out.println("sent " + bytesTransferred + " out of " + inputFile.length() + " in " + (et-st) + " millis");
        }

        socketChannel.finishConnect();
        long endTime = System.currentTimeMillis();

        System.out.println("sent: totalBytesTransferred= " + totalBytesTransferred + " / " + inputFile.length() + " in " + (endTime-startTime) + " millis");

        final File outputFile = new File(inputFile.getAbsolutePath() + ".out");
        boolean copyEqual = FileUtils.contentEquals(inputFile, outputFile);
        System.out.println("copyEqual= " + copyEqual);

        if (args.length > 1) {
            System.out.println("sleep: " + args[1] + " millis");
            Thread.sleep(Long.parseLong(args[1]));
        }
    }
}

服务器 - ZeroCopyServer.java:

import java.io.*;
import java.net.*;
import java.nio.channels.*;
import org.apache.commons.io.FileUtils;

public class ZeroCopyServer {

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

        final File inputFile = new File(args[0]);
        inputFile.delete();
        final File outputFile = new File(inputFile.getAbsolutePath() + ".out");
        outputFile.delete();

        createTempFile(inputFile, Long.parseLong(args[1])*1024L*1024L);

        System.out.println("input file length: " + inputFile.length() + " : output file.exists= " + outputFile.exists());

        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.socket().setReceiveBufferSize(8*1024*1024);
        System.out.println("server receive buffer size: " + serverSocketChannel.socket().getReceiveBufferSize());
        serverSocketChannel.socket().bind(new InetSocketAddress("localhost", 8083));
        System.out.println("waiting for connection");
        SocketChannel socketChannel = serverSocketChannel.accept();
        System.out.println("connected. client channel: " + socketChannel);

        FileOutputStream fileOutputStream = new FileOutputStream(outputFile);
        FileChannel fileChannel = fileOutputStream.getChannel();
        long startTime = System.currentTimeMillis();
        long transferFromByteCount = fileChannel.transferFrom(socketChannel, 0, inputFile.length());
        long endTime = System.currentTimeMillis();
        System.out.println("received: transferFromByteCount= " + transferFromByteCount + " : outputFile= " + outputFile.length() + " : inputFile= " + inputFile.length() + " bytes in " + (endTime-startTime) + " millis");

        boolean copyEqual = FileUtils.contentEquals(inputFile, outputFile);
        System.out.println("copyEqual= " + copyEqual);

        serverSocketChannel.close();

    }

    private static void createTempFile(File file, long size) throws IOException{
        RandomAccessFile f = new RandomAccessFile(file.getAbsolutePath(), "rw");
        f.setLength(size);
        f.writeDouble(Math.random());
        f.close();
    }

}

更新 1:Linux 代码修复了循环。

更新 2:我正在考虑的一种可能的解决方法需要客户端-服务器协作。在传输结束时，服务器将接收到的数据的长度写回客户端，客户端以阻塞模式读取它。

服务器响应:

ByteBuffer response = ByteBuffer.allocate(8);
response.putLong(transferFromByteCount);
response.flip();
socketChannel.write(response);   
serverSocketChannel.close(); 

客户端阻塞读取:

ByteBuffer response = ByteBuffer.allocate(8);
socketChannel.read(response);
response.flip();
long totalBytesReceived = response.getLong();

因此，客户端等待字节通过发送和接收套接字缓冲区，实际上等待字节存储在输出文件中。不需要实现带外确认，也不需要客户端按照 II.A 节中的建议等待 https://linuxnetworkstack.files.wordpress.com/2013/03/paper.pdf如果文件内容是可变的。

"wait an “appropriate” amount of time before rewriting the same portion of file"

更新 3:

包含@EJP 和@the8472 修复的修改示例，具有长度和文件校验和验证，没有输出跟踪。请注意，计算大型文件的 CRC32 校验和可能需要几秒钟才能完成。

客户:

import java.io.*;
import java.net.*;
import java.nio.*;
import java.nio.channels.*;
import org.apache.commons.io.FileUtils;

public class ZeroCopyClient {

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

        final File inputFile = new File(args[0]);

        FileInputStream fileInputStream = new FileInputStream(inputFile);
        FileChannel fileChannel = fileInputStream.getChannel();
        SocketAddress socketAddress = new InetSocketAddress("localhost", 8083);
        SocketChannel socketChannel = SocketChannel.open();
        socketChannel.connect(socketAddress);

        //send input file length and CRC32 checksum to server
        long checksumCRC32 = FileUtils.checksumCRC32(inputFile);
        ByteBuffer request = ByteBuffer.allocate(16);
        request.putLong(inputFile.length());
        request.putLong(checksumCRC32);
        request.flip();
        socketChannel.write(request);

        long totalBytesTransferred = 0;
        while (totalBytesTransferred < inputFile.length()) {
            long bytesTransferred = fileChannel.transferTo(totalBytesTransferred, inputFile.length()-totalBytesTransferred, socketChannel);
            totalBytesTransferred += bytesTransferred;
        }

        //receive output file length and CRC32 checksum from server
        ByteBuffer response = ByteBuffer.allocate(16);
        socketChannel.read(response);
        response.flip();
        long totalBytesReceived = response.getLong();
        long outChecksumCRC32 = response.getLong();

        socketChannel.finishConnect();

        System.out.println("CRC32 equal= " + (checksumCRC32 == outChecksumCRC32));

    }
}

服务器:

import java.io.*;
import java.net.*;
import java.nio.*;
import java.nio.channels.*;
import org.apache.commons.io.FileUtils;

public class ZeroCopyServer {

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

        final File outputFile = new File(args[0]);

        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.socket().bind(new InetSocketAddress(8083));     
        SocketChannel socketChannel = serverSocketChannel.accept();

        //read input file length and CRC32 checksum sent by client
        ByteBuffer request = ByteBuffer.allocate(16);
        socketChannel.read(request);
        request.flip();
        long length = request.getLong();
        long checksumCRC32 = request.getLong();

        FileOutputStream fileOutputStream = new FileOutputStream(outputFile);
        FileChannel fileChannel = fileOutputStream.getChannel();
        long totalBytesTransferFrom = 0;
        while (totalBytesTransferFrom < length) {
            long transferFromByteCount = fileChannel.transferFrom(socketChannel, totalBytesTransferFrom, length-totalBytesTransferFrom);
            if (transferFromByteCount <= 0){
                break;
            }
            totalBytesTransferFrom += transferFromByteCount;
        }

        long outChecksumCRC32 = FileUtils.checksumCRC32(outputFile);

        //write output file length and CRC32 checksum back to client
        ByteBuffer response = ByteBuffer.allocate(16);
        response.putLong(totalBytesTransferFrom);
        response.putLong(outChecksumCRC32);
        response.flip();
        socketChannel.write(response);

        serverSocketChannel.close();

        System.out.println("CRC32 equal= " + (checksumCRC32 == outChecksumCRC32));

    }
}

Answer 1

解决方案是检查 fileChannel.transferFrom 的写入计数器:

import java.io.*;
import java.net.*;
import java.nio.*;
import java.nio.channels.*;
import org.apache.commons.io.FileUtils;

public class ZeroCopyServer {

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

    final File outputFile = new File(args[0]);

    ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
    serverSocketChannel.socket().bind(new InetSocketAddress(8083));     
    SocketChannel socketChannel = serverSocketChannel.accept();

    //read input file length and CRC32 checksum sent by client
    ByteBuffer request = ByteBuffer.allocate(16);
    socketChannel.read(request);
    request.flip();
    long length = request.getLong();
    long checksumCRC32 = request.getLong();

    FileOutputStream fileOutputStream = new FileOutputStream(outputFile);
    FileChannel fileChannel = fileOutputStream.getChannel();
    long totalBytesTransferFrom = 0;
    while (totalBytesTransferFrom < length) {
        long transferFromByteCount = fileChannel.transferFrom(socketChannel, totalBytesTransferFrom, length-totalBytesTransferFrom);
        if (transferFromByteCount <= 0){
            break;
        }
        totalBytesTransferFrom += transferFromByteCount;
    }

    long outChecksumCRC32 = FileUtils.checksumCRC32(outputFile);

    //write output file length and CRC32 checksum back to client
    ByteBuffer response = ByteBuffer.allocate(16);
    response.putLong(totalBytesTransferFrom);
    response.putLong(outChecksumCRC32);
    response.flip();
    socketChannel.write(response);

    serverSocketChannel.close();

    System.out.println("CRC32 equal= " + (checksumCRC32 == outChecksumCRC32));

  }
}