java - 在 Java 输入/输出流上使用 BufferedReader 时 readline() 阻塞

Question

我正在制作一个 Java(版本控制)应用程序，其中客户端和服务器通过套接字流进行通信。我遇到的问题是客户端或服务器在执行 readline() 时往往会阻塞，最终导致另一端也阻塞。奇怪的是，第一次运行的时候好像没有出现这个错误。

我花了很多时间来找出导致问题的原因以及如何解决它，但我只能将我的问题强烈减少到下面提供的代码。 MAKE_STUB 参数允许使用管道流而不是套接字流。经常会导致线程在client读完之前就关闭导致的错误，但是有时候也会出现和使用sockets一样的错误，这大概说明问题不是socket的误用，而是stream的误用？

我还尝试了以下操作:更改刷新输出的时间，在 println() 中添加 +“\n”，依次发送/接收 (PING_PONG)，防止流被垃圾收集(使用静态列表)，发送一个字符串而不是 JSON 等。我觉得奇怪的是，这些解决方案只会导致 block 发生得更早或更晚。

客户端代码:

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.PipedInputStream;
import java.io.PipedOutputStream;
import java.io.PrintWriter;
import java.io.Writer;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.UnknownHostException;


public class TestClientProt {

private static final boolean MAKE_STUB = false;
private static final boolean PING_PONG = false;


private static void send(String value, InputStream rawInput, OutputStream rawOutput) throws IOException {
    PrintWriter output = new PrintWriter(rawOutput);
    output.println(value);
    output.flush();

    if(PING_PONG) {
        BufferedReader input = new BufferedReader(new InputStreamReader(rawInput));
        if(!input.readLine().equals(value))
            System.err.println("Ping pong failed!");
    }
}


private static int receiveInt(InputStream rawInput, OutputStream rawOutput) throws NumberFormatException, IOException {
    return Integer.parseInt(receiveString(rawInput, rawOutput));
}

private static String receiveString(InputStream rawInput, OutputStream rawOutput) throws IOException {
    BufferedReader input = new BufferedReader(new InputStreamReader(rawInput));
    String value = input.readLine();

    if(PING_PONG) {
        PrintWriter output = new PrintWriter(rawOutput);
        output.println(value);
        output.flush();
    }

    return value;
}


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

    InputStream rawInput;
    OutputStream rawOutput;

    if(MAKE_STUB) {

        rawInput = new PipedInputStream();
        final OutputStream rawServerOutput = new PipedOutputStream((PipedInputStream) rawInput);

        final PipedInputStream rawServerInput = new PipedInputStream();
        rawOutput = new PipedOutputStream(rawServerInput);

        new Thread() {

            public void run() {
                try {
                    TestServerProt.startSession(rawServerInput, rawServerOutput);
                } catch (IOException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }

        }.start();


    } else {

        InetAddress ip = InetAddress.getByName("127.0.0.1");
        int port = 6789;

        InetSocketAddress serverAddress = new InetSocketAddress(ip, port);      
        socket = new Socket();
        socket.connect(serverAddress);

        rawInput = socket.getInputStream();
        rawOutput = socket.getOutputStream();

    }


    System.out.println("start checkout");

    send("CHKT repname", rawInput, rawOutput);
    System.out.println(receiveInt(rawInput, rawOutput));

    send("RREV", rawInput, rawOutput);
    System.out.println(receiveInt(rawInput, rawOutput));
    System.out.println("receiving revision");
    System.out.println(receiveString(rawInput, rawOutput));
    System.out.println("revision received");

    System.out.println(receiveInt(rawInput, rawOutput));
    System.out.println("receiving files");

    send("RFIL", rawInput, rawOutput);
    System.out.println(receiveInt(rawInput, rawOutput));
    System.out.println("receiving file: ");
    System.out.println(receiveString(rawInput, rawOutput));

    System.out.println("success!");


    if(!MAKE_STUB)
        socket.close();
}

}

服务器代码:

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.PrintWriter;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;

import org.json.simple.JSONArray;
import org.json.simple.JSONObject;


public class TestServerProt {

private static final boolean PING_PONG = false;


public static void startSession(InputStream rawInput, OutputStream rawOutput) throws IOException {
    System.out.println("starting session");
    System.out.println(receiveString(rawInput, rawOutput));

    send(101, rawInput, rawOutput);

    System.out.println(receiveString(rawInput, rawOutput));
    System.out.println("sending revision");
    send(104, rawInput, rawOutput);
    send(createRepString(), rawInput, rawOutput);

    System.out.println("sending files");
    send(101, rawInput, rawOutput);
    System.out.println("sending files");

    System.out.println(receiveString(rawInput, rawOutput));
    send(103, rawInput, rawOutput);
    send("content file 1", rawInput, rawOutput);

    System.out.println("success!");

    //send(rCode);
}


private static void send(String value, InputStream rawInput, OutputStream rawOutput) throws IOException {
    PrintWriter output = new PrintWriter(rawOutput);
    output.println(value);
    output.flush();

    if(PING_PONG) {
        BufferedReader input = new BufferedReader(new InputStreamReader(rawInput));
        if(!input.readLine().equals(value))
            System.err.println("Ping pong failed!");
    }
}

private static void send(int value, InputStream rawInput, OutputStream rawOutput) throws IOException {
    send(String.valueOf(value), rawInput, rawOutput);
}


private static int receiveInt(InputStream rawInput, OutputStream rawOutput) throws NumberFormatException, IOException {
    return Integer.parseInt(receiveString(rawInput, rawOutput));
}

private static String receiveString(InputStream rawInput, OutputStream rawOutput) throws IOException {
    BufferedReader input = new BufferedReader(new InputStreamReader(rawInput));
    String value = input.readLine();

    if(PING_PONG) {
        PrintWriter output = new PrintWriter(rawOutput);
        output.println(value);
        output.flush();
    }

    return value;
}


public static void main(String[] args) throws IOException {
    int port = 6789;

    InetSocketAddress serverAddress = new InetSocketAddress(port);
    ServerSocket serverSocket = new ServerSocket();
    serverSocket.bind(serverAddress);

    System.out.println("waiting for client");

    Socket clientSocket = serverSocket.accept();

    System.out.println("client accepted");

    startSession(clientSocket.getInputStream(), clientSocket.getOutputStream());

    serverSocket.close();
    clientSocket.close();
}

private static String createRepString() {
    JSONObject json = new JSONObject();

    json.put("revision_nr", new Integer(5));

    JSONArray jsonArrayContent = new JSONArray();
    JSONObject jsonContent = new JSONObject();

    jsonContent.put("filename", "f1.txt");
    jsonContent.put("revision_nr", 3);
    jsonArrayContent.add(jsonContent);

    jsonContent.put("filename", "f2.txt");
    jsonContent.put("revision_nr", 1);
    jsonArrayContent.add(jsonContent);

    jsonContent.put("filename", "f3.txt");
    jsonContent.put("revision_nr", 5);
    jsonArrayContent.add(jsonContent);

    json.put("content", jsonArrayContent);

    JSONArray jsonArrayStatus = new JSONArray();
    JSONObject jsonStatus = new JSONObject();

    json.put("status", jsonArrayStatus);

    return json.toJSONString();
}

}

Answer 1

在您的情况下，最重要的问题是您每次阅读时都会创建一个新的 BufferedReader 。问题是 BufferedReader 将用可用数据填满它的整个缓冲区——因为它应该这样做。因此它将从底层流读取超出行尾的部分。

解决方案很简单:只创建一次 BufferedReader，然后传递它，而不是传递 InputStream。