java - 交替使用 PipedOutputStream PipedInputStream Numbers 示例

Question

我想使用 PipedOutputStream 和 PipedInputStream Java 类编写类似于Producer Consumer Problem 的示例。

注意:这是应用其原理的小示例。

public static void main(String[] args) {
  try {
    final PipedOutputStream pipedSrc = new PipedOutputStream();
    final PipedInputStream pipedSnk = new PipedInputStream();
    pipedSnk.connect(pipedSrc);
    int putNumbers = ThreadLocalRandom.current().nextInt(5, 10);

    Runnable runnableGet = () -> {
      int getNumbers = getNumbers(pipedSnk);
      System.out.println("Read " + getNumbers + " numbers from the PipedInputStream");
    };
    Thread threadGet = new Thread(runnableGet, "threadGet");
    threadGet.start();

    Runnable runnablePut = () -> {
      System.out.println("Write: " + putNumbers + " numbers to the PipedOutputStream");
      putNumbers(pipedSrc, putNumbers);
    };
    Thread threadPut = new Thread(runnablePut, "threadPut");
    threadPut.start();

  } catch (IOException e) { /*Ignored Now*/}
}

现在是方法:

public static void putNumbers(PipedOutputStream pipedSrc, int numbers) {
  try {
      for (int i = 0; i < numbers; i++) {
        Integer number = ThreadLocalRandom.current().nextInt(0, 100);
        System.out.println("Put number: " + number);
        pipedSrc.write(number.toString().getBytes());
        try {
          int timeSleeping = ThreadLocalRandom.current().nextInt(500, 1000);//Simulate some duration
          Thread.sleep(timeSleeping);
        } catch (InterruptedException ex) {
          System.out.println(ex.toString());
        }
      }
      pipedSrc.flush();
      pipedSrc.close();
  } catch (IOException e) { /*Ignored Now*/}
}

public static int getNumbers(PipedInputStream pipedSnk) {
  int numbers = 0;
  try {
      byte[] readBytes = new byte[8];
      int qtyBytes;
      while ((qtyBytes = pipedSnk.read(readBytes)) != -1) {
        numbers++;
        byte[] bytesNumber = new byte[qtyBytes];
        System.arraycopy(readBytes, 0, bytesNumber, 0, qtyBytes);
        System.out.println("Get Number: " + new String(bytesNumber));
      }
  } catch (IOException e) { /*Ignored Now*/}
  return numbers;
}

输出 1(混合):

$ java -cp /.../Java/classes PipedNumbers
Write: 7 numbers to the PipedOutputStream
Put number: 36
Put number: 92
Get Number: 3692
Put number: 22
Put number: 55
Get Number: 2255
Put number: 64
Get Number: 64
Put number: 2
Get Number: 2
Put number: 82
Get Number: 82
Read 5 numbers from the PipedInputStream
$ 

更改源代码。

public static void putNumbers(PipedOutputStream pipedSrc, int numbers) {
  try {
    DataOutputStream dos = new DataOutputStream(new BufferedOutputStream(pipedSrc));
    for (int i = 0; i < numbers; i++) {
      Integer number = ThreadLocalRandom.current().nextInt(0, 100);
      System.out.println("Put number: " + number);
      dos.writeInt(number);
      try {
        int timeSleeping = ThreadLocalRandom.current().nextInt(500, 1000);//Simulate some duration
        Thread.sleep(timeSleeping);
      } catch (InterruptedException ex) {
        System.out.println(ex.toString());
      }
    }
    dos.flush();
    dos.close();
  } catch (IOException e) { /*Ignored Now*/}
}

public static int getNumbers(PipedInputStream pipedSnk) {
  int numbers = 0;
  try {
    DataInputStream dis = new DataInputStream(new BufferedInputStream(pipedSnk));
    Boolean end = false;
    while (!end) {
      try {
        Integer number = dis.readInt();
        System.out.println("Get Number: " + number);
        numbers++;
      } catch (EOFException e) {
        end = true;
      }
    }
  } catch (IOException e) { /*Ignored Now*/}
  return numbers;
}

输出 2(不是隔行扫描/交替):

$ java -cp /.../Java/classes PipedNumbers
Write: 7 numbers to the PipedOutputStream
Put number: 17
Put number: 70
Put number: 88
Put number: 12
Put number: 60
Put number: 19
Put number: 41
Get Number: 17
Get Number: 70
Get Number: 88
Get Number: 12
Get Number: 60
Get Number: 19
Get Number: 41
Read 7 numbers from the PipedInputStream
$ 

如何使用 PipedInputStream 和 PipedOutputStream 交错和分离生成的数字？

Answer 1

只需在每次迭代后调用 dos.flush() 而不是在最后调用一次。

发生这种情况是因为您的 DataOutputStream 具有 BufferedOutputStream，因此它会在写入输出流之前缓冲您的 PipedOutputStream。如果你想在每个 dos.writeInt() 语句之后总是立即写入而不需要缓冲调用 dos.flush() 或者不要使用缓冲流。

所以我的 putNumbers 看起来像这样:

public static void putNumbers(PipedOutputStream pipedSrc, int numbers) {
    try {
        DataOutputStream dos = new DataOutputStream(pipedSrc);
        for (int i = 0; i < numbers; i++) {
            Integer number = ThreadLocalRandom.current().nextInt(0, 100);
            System.out.println("Put number: " + number);
            dos.writeInt(number);
            try {
                int timeSleeping = ThreadLocalRandom.current().nextInt(500, 1000);//Simulate some duration
                Thread.sleep(timeSleeping);
            } catch (InterruptedException ex) {
                System.out.println(ex.toString());
            }
            dos.flush(); //<--changed
        }
        dos.close();
    } catch (IOException e) { /*Ignored Now*/}
}