java - 为什么我在尝试读取 DataInputStream 时收到 EOFException？

Question

我正在做一项作业，我必须使用 Java Socket 和 Thread 制作一个聊天程序，但我无法让它工作。我从 DataInputStream 获取 EOFEception。据我了解，DataInputStream 中的 EOFException 意味着流已关闭，但我尚未关闭该流。

编辑:我和我的教授一起坐了一个小时，但她也无法弄清楚，所以对任何能弄清楚的人表示敬意。

聊天.java

package dev.gigaboy.chat;

import java.io.IOException;
import java.net.Socket;

import javafx.application.Application;
import javafx.application.Platform;
import javafx.beans.value.ChangeListener;
import javafx.beans.value.ObservableValue;
import javafx.event.ActionEvent;
import javafx.event.EventHandler;
import javafx.geometry.Insets;
import javafx.geometry.Rectangle2D;
import javafx.scene.Scene;
import javafx.scene.control.Button;
import javafx.scene.control.TextArea;
import javafx.scene.control.TextField;
import javafx.scene.input.KeyCode;
import javafx.scene.layout.HBox;
import javafx.scene.layout.VBox;
import javafx.scene.text.Font;
import javafx.stage.Screen;
import javafx.stage.Stage;

public class Chat extends Application {

    private static final int WIDTH = 600, HEIGHT = 290;

    private static String name;
    private static IOThread buffThread;

    private static Insets ins = new Insets(2.5, 5, 2.5, 5);
    private static Font fnt = new Font("Calibri", 13);
    public static TextArea chatBox, userBox;

    public static void main(String[] args) {
        launch(args);
    }

    @Override
    public void start(Stage prmStage) throws Exception {
        ////////// CHAT SCENE //////////
        HBox hbox = new HBox();

        // current message
        TextField tf = new TextField();
        tf.setOnKeyPressed(e -> {
            if (e.getCode() == KeyCode.ENTER) {
                try {
                    buffThread.getWriter().writeUTF(name + ": " + tf.getText());
                    buffThread.getWriter().flush();
                } catch (IOException e1) {
                    e1.printStackTrace();
                }
                System.out.println("Buffer Flushed");

                // append message and scroll TextArea to bottom
                chatBox.appendText(name + ": " + tf.getText() + "\n");
                chatBox.selectPositionCaret(chatBox.getLength());
                chatBox.deselect();
                tf.clear();
            }
        });

        // box where messages show up
        chatBox = new TextArea();
        chatBox.setWrapText(true);
        chatBox.setEditable(false);

        // box where users show up
        userBox = new TextArea();
        userBox.setMinSize(140, 250);
        userBox.setMaxSize(140, 250);
        userBox.setEditable(false);

        hbox.getChildren().addAll(chatBox, userBox);

        VBox vbox = new VBox();
        vbox.getChildren().addAll(hbox, tf);
        HBox.setMargin(chatBox, ins);
        HBox.setMargin(userBox, ins);
        VBox.setMargin(tf, ins);

        ////////// CONNECTION SCENE //////////
        VBox connBox = new VBox();

        TextField usr = new TextField("Username");
        usr.setFont(fnt);

        TextField host = new TextField("localhost");
        host.setFont(fnt);

        // set port to only allow positive integers
        TextField port = new TextField("7777");
        port.setFont(fnt);
        port.textProperty().addListener(new ChangeListener<String>() {
            @Override
            public void changed(ObservableValue<? extends String> observable, String oldValue, String newValue) {
                if (!newValue.matches("\\d*")) {
                    port.setText(newValue.replaceAll("[^\\d]", ""));
                }
            }
        });

        Button bttn = new Button("Connect");
        bttn.setFont(fnt);
        bttn.setOnAction(new EventHandler<ActionEvent>() {
            @Override
            public void handle(ActionEvent evt) {
                // connect to server or become server
                try {
                    connect(host.getText(), Integer.parseInt(port.getText()));
                } catch (NumberFormatException | IOException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }

                prmStage.setScene(new Scene(vbox, WIDTH, HEIGHT));
                prmStage.setTitle(usr.getText() + " at " + host.getText() + ":" + port.getText());
                name = usr.getText();

                // reposition window
                Rectangle2D scrnBounds = Screen.getPrimary().getVisualBounds();
                prmStage.setX((scrnBounds.getWidth() - WIDTH) / 2);
                prmStage.setY((scrnBounds.getHeight() - HEIGHT) / 2 - 50);
            }
        });

        connBox.getChildren().addAll(usr,host, port, bttn);
        VBox.setMargin(usr, ins);
        VBox.setMargin(host, ins);
        VBox.setMargin(port, ins);
        VBox.setMargin(bttn, ins);


        Scene connScene = new Scene(connBox, 230, 130);
        prmStage.setOnCloseRequest(evt -> {
            try {
                if (buffThread != null)
                    buffThread.close();
            } catch (IOException e1) {
                // TODO Auto-generated catch block
                e1.printStackTrace();
            }
            Platform.exit();
            System.exit(0);
        });
        prmStage.setTitle("Connect...");
        prmStage.setScene(connScene);
        prmStage.setResizable(false);
        prmStage.show();
    }

    private static boolean isHostAvailable(String host, int port) {
        try (Socket socket = new Socket(host, port)) {
            return true;
        } catch (IOException e) {

        }

        return false;
    }

    private static void connect(String host, int port) throws IOException {
        if (isHostAvailable(host, port)) {
            buffThread = new ClientThread(host, port);
            System.out.println("CLIENT");
        } else {
            buffThread = new ServerThread(port);
            System.out.println("SERVER");
        }

        buffThread.start();
    }

}

ClientThread.java

package dev.gigaboy.chat;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.Socket;

import javafx.application.Platform;

public class ClientThread extends IOThread {

    private Socket socket;

    private String host;
    private int port;

    public ClientThread(String host, int port) {
        this.host = host;
        this.port = port;
    }

    @Override
    public void close() throws IOException {
        socket.close();
        super.close();
    }

    @Override
    public void run() {
        try {
            socket = new Socket(host, port);
            // inherited from BufferedThread
            in = new DataInputStream(socket.getInputStream());
            out = new DataOutputStream(socket.getOutputStream());
        } catch (IOException e1) {
            e1.printStackTrace();
        }

        while (true) {
            Platform.runLater(() -> {
                try {
                    String line;
                    line = in.readUTF();
                    Chat.chatBox.appendText(line);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            });
        }
    }

}

ServerThread.java

package dev.gigaboy.chat;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;

import javafx.application.Platform;

public class ServerThread extends IOThread {

    private ServerSocket server;
    private Socket socket;

    private int port;

    public ServerThread(int port) {
        super();
        this.port = port;
    }

    @Override
    public void close() throws IOException {
        server.close();
        socket.close();
        super.close();
    }

    @Override
    public void run() {
        try {
            server = new ServerSocket(port);
            socket = server.accept();

            // inherited from BufferedThread
            in = new DataInputStream(socket.getInputStream());
            out = new DataOutputStream(socket.getOutputStream());
        } catch (IOException e1) {
            e1.printStackTrace();
        }

        while (true) {
            Platform.runLater(() -> {
                try {
                    String line;
                    line = in.readUTF();
                    Chat.chatBox.appendText(line);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            });
        }
    }

}

IOThread.java

package dev.gigaboy.chat;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;

public class IOThread extends Thread {

    protected DataInputStream in;
    protected DataOutputStream out;

    public IOThread() {

    }

    public DataInputStream getReader() {
        return in;
    }

    public DataOutputStream getWriter() {
        return out;
    }

    public String readUTF() throws IOException {
        return in.readUTF();
    }

    public void close() throws IOException {
        in.close();
        out.close();
    }

}

Answer 1

我认为 readUTF 可能不理解它应该等待行字符的结尾，这就是发生这种情况的原因。

使用 DataFetcher 和 FetcherListener ( https://sourceforge.net/p/tus/code/HEAD/tree/tjacobs/io ) 可以让您的生活变得更轻松

package tjacobs.io;

import java.io.IOException;
import java.io.InputStream;
import java.io.PrintStream;
import java.util.ArrayList;
import java.util.Iterator;

import tjacobs.io.TimeOut.TimeOutCmd;

/**
 * DataFetcher is a generic way to read data from an input stream (file, socket, etc)
 * DataFetcher can be set up with a thread so that it reads from an input stream 
 * and report to registered listeners as it gets
 * more information. This vastly simplifies the process of always re-writing
 * the same code for reading from an input stream.
 * @author tjacobs01
 */
    public class DataFetcher implements Runnable, TimeOutCmd {
        public byte[] buf;
        public InputStream in;
        public int waitTime;
        private ArrayList<FetcherListener> mFetcherListeners;
        public int got = 0;
        protected boolean mClearBufferFlag = false;

        public DataFetcher(InputStream in, byte[] buf, int waitTime) {
            this.buf = buf;
            this.in = in;
            this.waitTime = waitTime;
        }

        public void addFetcherListener(FetcherListener listener) {
            if (mFetcherListeners == null) {
                mFetcherListeners = new ArrayList<FetcherListener>(2);
            }
            if (!mFetcherListeners.contains(listener)) {
                mFetcherListeners.add(listener);
            }
        }

        public void removeFetcherListener(FetcherListener fll) {
            if (mFetcherListeners == null) {
                return;
            }
            mFetcherListeners.remove(fll);
        }

        public byte[] readCompletely() {
            run();
            return buf;
        }

        public int got() {
            return got;
        }

        /** Override this to implement other implementations
         * 
         */
        public void timeOut() {
            try {
                if (in != null)
                in.close();
            }
            catch (IOException iox) {
                iox.printStackTrace();
            }
        }

        public void run() {
            TimeOut to = null;
            if (waitTime > 0) {
                to = new TimeOut(this, waitTime);
                Thread t = new Thread(to);
                t.start();
            }           
            int b;
            try {
                if (in == null) {
                    signalListeners(true);
                    return;
                }
                while ((b = in.read()) != -1) {
                    if (to != null) to.tick();
                    if (got + 1 > buf.length) {
                        buf = IOUtils.expandBuf(buf);
                    }
                    int start = got;
                    buf[got++] = (byte) b;
                    int available = in.available();
                    //System.out.println("got = " + got + " available = " + available + " buf.length = " + buf.length);
                    if (got + available > buf.length) {
                        buf = IOUtils.expandBuf(buf, Math.max(got + available, buf.length * 2));
                    }
                    got += in.read(buf, got, available);
                    signalListeners(false, start);
                    if (mClearBufferFlag) {
                        mClearBufferFlag = false;
                        got = 0;
                    }
                }
            } catch (IOException iox) {
                iox.printStackTrace();
                throw new PartialReadException(got, buf.length);
            } finally {
                if (to != null) to.stop();
                buf = IOUtils.trimBuf(buf, got);
                signalListeners(true);
            }
        }

        private void setClearBufferFlag(boolean status) {
            mClearBufferFlag = status;
        }

        public void clearBuffer() {
            setClearBufferFlag(true);
        }

        private void signalListeners(boolean over) {
            signalListeners (over, 0);
        }

        private void signalListeners(boolean over, int start) {
            if (mFetcherListeners != null) {
                Iterator<FetcherListener> i = mFetcherListeners.iterator();
                while (i.hasNext()) {
                    FetcherListener fll = i.next();
                    if (over) {
                        fll.fetchedAll(buf);
                    } else {
                        fll.fetchedMore(buf, start, got);
                    }
                }               
            }
        }

        public static interface FetcherListener {
            public void fetchedMore(byte[] buf, int start, int end);
            public void fetchedAll(byte[] buf);
        }

        public static class ToPrintStream implements FetcherListener {
            PrintStream stream;
            public ToPrintStream(PrintStream ps) {
                stream = ps;
            }

            @Override
            public void fetchedAll(byte[] buf) {
            }

            @Override
            public void fetchedMore(byte[] buf, int start, int end) {
                stream.print(new String(buf, start, end - start));
            }           
        }

        public static class ToStandardOut extends ToPrintStream {

            public ToStandardOut() {
                super(System.out);
            }
        }       
    }