通过管道发送对象

Question

我是 Java 的初学者，也许我想做一些在 Java 中不起作用的东西，但我希望这里有人可以帮助我。

我想做什么： 我有两个线程，一个线程创建几个相同的对象并使用它，最后我想通过管道将这些对象发送到另一个线程，我想将它保存在一个文件中。

我知道，我可以使用 Vector，因此其他线程可以从那里获取对象，但是在下一步中，我想在服务器和客户端中分离这个项目，所以我需要这些管道来发送对象。我已经在互联网上搜索了正确的答案，但我确实找到了我需要的任何东西。这是我认为可行但没有实现的想法：

主线程：

public class ControlerThread {

public static void main(String[] args) {
    PipedReader pr = new PipedReader();
    PipedWriter pw = new PipedWriter();


    PipedOutputStream pos =  null;
    PipedInputStream pis = null;
    ObjectOutputStream oos =null;
    ObjectInputStream ois = null;
    try
    {    
        pw.connect(pr);
        pis =  new PipedInputStream();
        pos = new PipedOutputStream(pis);
        oos = new ObjectOutputStream(pos);
        ois = new ObjectInputStream(pis);

    }catch(IOException ioe)
    {

    }
        ModelThread mt = new ModelThread(ois,pr);
        ViewThread vt = new ViewThread(oos,pw);

        vt.start();
        mt.start();
 }
}

我在其中创建对象的 ViewThread...

public class ViewThread extends Thread{
PipedWriter pw;
BufferedWriter bw;

ObjectOutputStream oos;
public ViewThread(ObjectOutputStream oos,PipedWriter pwr )
{
    this.oos = oos;
    this.pw = pwr;
    bw = new BufferedWriter(pw);

}
public void run()
{
    int iEingabe = 0;
    Menu mu = new Menu();
    do {

        iEingabe = 0;

        System.out.println("\t  eine neue Person aufnehmen: > 1");
        System.out.println("\t \t   Records auflisten: > 2");
        System.out.println("       Records in eine Datei sichern: > 3");
        System.out.println("       Records aus einer Datei laden: > 4");
        System.out.println("         in-memory Records sortieren: > 5");
        System.out.println("\t\t       Datei löschen: > 6");
        System.out.println("\t      das Programm verlassen: > 7");

        System.out.print("\nIhre Eingabe: ");           
        iEingabe = Eingabe.readInt();

        switch (iEingabe) {
        case 1: mu.addContact(); break;

        case 2: mu.outputMatrix(); break;

        case 3: try{

                bw.write("3");
                bw.write("\n");
                bw.flush();

                Kontakt k = new Kontakt();

                oos.writeObject(k);
                oos.flush();

            }catch(IOException ioe)
            {
                ioe.printStackTrace();
            }; break;

        case 4:  break;
        case 5: mu.kontakteSortieren();break;

        case 6: /*mu.deleteFile()*/; break;

        case 7: System.out.println("Programm wird beended!"); break;
        }
    } while (iEingabe != 7);
}
}

这是稍后应该获取对象的线程：

package Thread;
import java.io.*;
public class ModelThread  extends Thread{

private PipedReader pr = null;
private BufferedReader br = null;
//private String datei = "test.csv";
//private FileOutputStream fos;
//private ObjectOutputStream oos = null;

//  private FileInputStream fis;
private ObjectInputStream ois = null;

public ModelThread(ObjectInputStream ois,PipedReader pr)
{
    this.pr = pr;
    br = new BufferedReader(pr);
    this.ois = ois;
}

public void run()
{
    try
    { 
        if(br.readLine().equals("3"))
        {
            Kontakt k = (Kontakt) ois.readObject();
            System.out.println(k.name);
        }
    }catch(IOException e)
    {

    }catch(ClassNotFoundException ce)
    {
    }
}
}

但是如果我像上面那样做，我会得到一个 IOException “Read end dead”。我希望每个人都明白我想做什么并能帮助我。

例外：

java.io.IOException: Read end dead
at java.io.PipedInputStream.checkStateForReceive(Unknown Source)
at java.io.PipedInputStream.receive(Unknown Source)
at java.io.PipedOutputStream.write(Unknown Source)
at java.io.ObjectOutputStream$BlockDataOutputStream.writeBlockHeader(Unknown Source)
at java.io.ObjectOutputStream$BlockDataOutputStream.drain(Unknown Source)
at java.io.ObjectOutputStream$BlockDataOutputStream.flush(Unknown Source)
at java.io.ObjectOutputStream.flush(Unknown Source)
at Thread.ViewThread.run(ViewThread.java:57)

谢谢！！！

Answer 1

在此 ModelThread 处理命令他侦听的数据之前，您的 ViewThread 将对象数据发送到 ModelThread 肯定存在竞争条件。

发生了什么

java.io.IOException: Read end dead 表示您的PipedInputStream 实际上还没有准备好工作。查看源代码，当管道输入流“没有人”（实际上没有线程）从中读取时，会抛出这个特定的异常。那么让我们看看您的代码何时开始从 PipedInputStream 读取数据。

看到你的代码（我可以建议你重命名你的变量吗？从一个陌生人的角度来看，真的很难理解为什么每个类都需要这么多的流，而当他们的名字像 pis , ois, pr...)，我们有 2 个线程共享一个“命令”通道（即读取器/写入器和一个交换序列化数据的“对象”（或“数据”）通道。

一旦您的程序启动，当用户输入“3”（在 System.in 中？）时，编写器线程：将“3”写入命令管道，然后将对象写入对象管道（这是您崩溃的地方)。
当其他线程“看到”命令管道上的 3 时，它开始从数据管道监听（我们知道这没有发生，因为我们崩溃了）。

为什么在 ViewThread 开始写入数据管道之前 ModelThread 没有开始监听数据管道？好吧，因为你的程序需要...

任务同步

最坏的情况：您甚至无法确定您的 ModelThread 是否真正启动（当然，您可以预期它已经启动，但您无法知道）。如果你不知道它已经开始，你肯定不知道它是否在监听命令管道，更不用说它从命令管道中读取以及它已经从对象管道中继续读取了。

嗯.. 实际上：考虑到您在对象管道上的异常，我们可以知道在您的程序的这个实例中，命令管道工作，它意味着这两个线程实际上都已启动，并且您的 ModelThread 至少进入了命令管道的读取方法。但你无法知道下次启动程序时会出现这种情况。

无论如何，在这种情况下，命令管道正常工作，您将命令“3”向下发送。接下来是什么。

你的 ModelThread最终会收到命令“3”，一段时间后，即使只是几分之一秒，也会开始从对象中读取管道......但话又说回来：你不知道这什么时候会发生。在 ViewThread 开始写下对象管道之前或之后？如果它之前：您的程序应该可以工作。如果没有，你会得到你提到的异常。

您的崩溃场景

嗯，总有一点未知，但在全球范围内

1. 两个线程都在等待。用户类型“3”
2. 操作系统决定将此输入信号发送给 ViewThread 以处理 3。
3. ViewThread 唤醒、运行、读取 3 命令并将其发送到管道中
4. OS 做出反应并发出信号 ModelThread 去（或者它可能还没有反应，......你的里程可能会有所不同）
5. 但是您有一个 4 核，与此同时，仍在运行的 ViewThread 已开始记录对象管道。崩溃。

第 4 步至关重要。无论您是否有多个 CPU，无论您的计算机是否超载，您都无法知道调度程序会选择做什么以及相对于您的读取时间它会走多快。

继续前进

嗯，这就是我们发明同步工具的原因。在您当前的设计中，您需要同步 (CountDownLatch?) 以确保至少您的阅读方面已经在您的写作方面开始。但从长远来看，这可能并不理想......

看到您的目标是客户端/服务器应用程序，这不会：同步工具（锁、信号量、任何类型的信号）在一个进程（或最多一台计算机）内工作，但您不会能够在客户端/服务器架构上与他们合作。

另外：在这个架构中你可能不会有管道，你会有普通的 InputStream / Output 流。所以我不确定你是否通过这种方式设计让你的设计变得更简单。

这实际上是一个机会（某种）：网络通信涉及套接字。套接字是双向通信支持：您可以读取和写入套接字。所以也许你可以修改你的协议，以便：

1. 你开始一个“会话”
2. 一方发送命令并等待确认
3. 对方接收命令并发送确认
4. 您继续发送实际数据（通过同一条线路，通过另一条线路，无论如何）

确认阶段是您的“同步”工具。当你收到它时，就意味着对方已经“准备好了”。

旁注

对于它的价值：您的“协议”让我想起了旧的 FTP...您有一个“命令通道”（客户端要求服务器列出目录、请求文件等），并且当必须进行传输时，客户端和服务器会为此数据交换打开第二个连接（数据连接）。

不是说你应该这样做，但它可能会激励你。