java - 流关闭错误 - Storm

Question

我试图在喷嘴中读取文件中的一行，然后将其发送到 bolt ，但我不断收到流关闭错误。是我关闭错误还是这里出了什么问题？

public class InputSpout extends BaseRichSpout {
    private SpoutOutputCollector collector;
    public void
    declareOutputFields( OutputFieldsDeclarer declarer) {
        declarer.declare( new Fields("logfile"));
    }
    private FileReader fileReader;
    private boolean completed = false;
    private TopologyContext context;
    @Override
    public void open( Map config, TopologyContext context, SpoutOutputCollector collector) { 
        try {
            this.context = context;
            this.fileReader = new FileReader(("logfile.txt").toString());
        } catch (FileNotFoundException e) {
            throw new RuntimeException("Error reading file "
                    + ("logfile"));
        }
        this.collector = collector;
    }
    public void nextTuple() {


        if (completed) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {

            }
        }
        String str;
        BufferedReader reader = new BufferedReader(fileReader);
        str = null;
        try {
            str = reader.readLine();
        } catch (IOException e1) {
            e1.printStackTrace();
        }
        try {
            while (str != null) {
                this.collector.emit(new Values(str));
                str = reader.readLine();
            }

        } catch (Exception e) {
            throw new RuntimeException("Error reading typle", e);
        } finally {
            completed = true;
        }
        try {
            reader.close();
            fileReader.close();
        } catch (IOException e) {
            e.printStackTrace();
        }

这是我收到的错误:

java.io.IOException: Stream closed at sun.nio.cs.StreamDecoder.ensureOpen(StreamDecoder.java:46) at sun.nio.cs.StreamDecoder.read(StreamDecoder.java:147) at java.io.InputStreamReader.read(InputStreamReader.java:184) at java.io.BufferedReader.fill(BufferedReader.java:154) at java.io.BufferedReader.readLine(BufferedReader.java:317) at java.io.BufferedReader.readLine(BufferedReader.java:382) at myStorm.InputSpout.nextTuple(InputSpout.java:52) at backtype.storm.daemon.executor$fn__4654$fn__4669$fn__4698.invoke(executor.clj:565) at backtype.storm.util$async_loop$fn__458.invoke(util.clj:463) at clojure.lang.AFn.run(AFn.java:24) at java.lang.Thread.run(Thread.java:745)

Answer 1

您将关闭 nextTuple 中 fileReader 使用的 InputStream，使其无法用于后续调用。无需关闭此 Reader - 只需在读取所有数据后关闭单个 BufferedReader 即可。