java - IOException - read end dead - 在此示例中导致它的原因以及如何修复它 - Java 中的多线程应用程序

Question

这是一个 extension我的问题发布在这里，虽然这似乎解决了我的问题的一部分，但现在我看到了 IO-Exception read end dead exception。我正在使用 多线程 应用程序，其中 thread-1 产生 随机数，其他 thread-2 使用 它来计算平均值。一旦平均值达到阈值，我就会向 thread-1 发出停止生成数字的信号。这是代码的基本设计。

我收到 IO-Exception read end dead exception。我想知道它为什么会出现以及如何解决它。 谢谢。

下面的代码:

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.PipedInputStream;
import java.io.PipedOutputStream;
import java.util.Random;
import java.util.concurrent.atomic.AtomicBoolean;

//

class NumGen extends Thread {

    PipedOutputStream pos;
    DataOutputStream dos;
    AtomicBoolean isDone;

    public NumGen(PipedOutputStream pos,AtomicBoolean isDone){
        this.pos=pos;
        dos=new DataOutputStream(pos);
        this.isDone=isDone;
    }

    public void run(){
        while (!isDone.get()){
            Random rand = new Random();
            try {
                dos.writeDouble(rand.nextDouble()+100.0);
            } catch (IOException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }
}

class RunningAvg extends Thread {

    PipedInputStream pis;
    DataInputStream dis;
    Double avg;
    int count;
    Double runningTotal;
    AtomicBoolean isDone;

    public RunningAvg(PipedInputStream pis,AtomicBoolean isDone){
        this.pis=pis;
        dis=new DataInputStream(pis);
        runningTotal=0.0;
        avg=0.0;
        this.isDone=isDone;
    }

    public void run(){
        try {
        while (dis.available()>0){
            count+=1;
            runningTotal+=dis.readDouble();
                avg=runningTotal/count;
                System.out.printf("The average in count no : %s is %s%n",count,avg);
                if (avg>1E5)
                  isDone.set(true);
        }
            } catch (IOException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }


public class InterThreadComm {

    public static void main(String[] args){


    try {
        PipedOutputStream pos= new PipedOutputStream();
        PipedInputStream pis = new PipedInputStream(pos);
        AtomicBoolean isDone = new AtomicBoolean(false);
        NumGen ng = new NumGen(pos,isDone);
        RunningAvg ra = new RunningAvg(pis,isDone);
        ng.start();
        ra.start();
    } catch (IOException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }
    }


}

编辑: 根据以下答案:我尝试使用 try-with-resources 关闭 Streams。但我明白了 java.io.IOException:管道关闭

public class InterThreadComm {

    public static void main(String[] args){


    try(PipedOutputStream pos= new PipedOutputStream();PipedInputStream pis =new PipedInputStream(pos)){ 
        AtomicBoolean isDone = new AtomicBoolean(false);
        NumGen ng = new NumGen(pos,isDone);
        RunningAvg ra = new RunningAvg(pis,isDone);
        ng.start();
        ra.start();
    } catch (IOException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }
    }


}

Answer 1

根本原因在这里:

while (dis.available()>0){

发生的情况是消费者线程有时能够使用所有数据。没有可用的，所以它很早就坏了。

抛出异常是因为 PipedInputStream/PipedOutputStream 跟踪正在读取和写入的线程。有一个名为 checkStateForReceive 的私有(private)方法抛出，基本上它是在提示你的消费者线程已经结束:

} else if (readSide != null && !readSide.isAlive()) {
    throw new IOException("Read end dead");
}

(readSide 是消费者线程。)

您可以在堆栈跟踪中看到它被调用:

java.io.IOException: Read end dead
    at java.io.PipedInputStream.checkStateForReceive(PipedInputStream.java:246)
    at java.io.PipedInputStream.receive(PipedInputStream.java:210)
    at java.io.PipedOutputStream.write(PipedOutputStream.java:132)
    at java.io.DataOutputStream.writeLong(DataOutputStream.java:207)
    at java.io.DataOutputStream.writeDouble(DataOutputStream.java:242)

So write calls receieve on the input stream, which calls checkStateForReceieve and throws if the reading Thread isn't alive.

I think your loop condition should just be:

while(!isDone.get()) {

与异常无关，我还认为你可能在这里有一个本质上的无限循环:

if (avg>1E5)
    isDone.set(true);

您的数字生成器生成 100 到 101 之间的数字，因此对它们进行平均永远不会大于 1e5。您可能想检查 runningTotal > 1E5。

还有:

• 不要忘记在完成后关闭流。
• 您的消费者线程在其循环外捕获 IOException，但生产者在循环内捕获。如果出现异常，生产者不会中止:它只是继续运行，一遍又一遍地捕获异常。您可能希望像消费者一样将 catch 移到循环之外。

关于您的编辑:

如果你想关闭 main 中的流，你可以在线程上join。

try(
    PipedOutputStream pos= new PipedOutputStream();
    PipedInputStream pis =new PipedInputStream(pos)
){ 
    AtomicBoolean isDone = new AtomicBoolean(false);
    NumGen ng = new NumGen(pos,isDone);
    RunningAvg ra = new RunningAvg(pis,isDone);

    ng.start();
    ra.start();

    try {
       ng.join(); // wait for ng and ra to complete
       ra.join(); //
    } catch(InterruptedException ie) {}
} catch (IOException e) {
    e.printStackTrace();
}