поочередно PipedOutputStream Пример чисел PipedInputStream

Question

Я хочу написать аналогично Пример проблемы с источником с использованием PipedOutputStream и PipedInputStream классов Java.

ПРИМЕЧАНИЕ. Это небольшой пример применения принципа.

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

public static void putNumbers(PipedOutputStream pipedSrc, int numbers) {
  try {
      byte[] writeBytes;
      for (int i = 0; i < numbers; i++) {
        Integer number = ThreadLocalRandom.current().nextInt(0, 1000);
        System.out.println("Put number: " + number);
        byte[] bytesNumber = number.toString().getBytes();
        writeBytes = new byte[8];
        System.arraycopy(bytesNumber, 0, writeBytes, 0, bytesNumber.length);
        pipedSrc.write(writeBytes);
        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*/}
}

выход:

Write: 8 numbers to the PipedOutputStream
Put number: 25
Put number: 57
Get Number: 25
Get Number: 57
Put number: 37
Get Number: 37
Put number: 31
Get Number: 31
Put number: 36
Put number: 10
Get Number: 36
Get Number: 10
Put number: 86
Get Number: 86
Put number: 59
Get Number: 59
Read 8 numbers from the PipedInputStream

Answer 2

Просто вызывайте dos.flush() после каждой итерации, а не один раз в конце.

Это происходит потому, что у вашего DataOutputStream есть BufferedOutputStream, поэтому он буферизует перед записью в выходной поток вашего PipedOutputStream.Если вы хотите всегда сразу писать без буферизации, звоните dos.flush() после каждого оператора dos.writeInt() или не используйте буферизованный поток.

Так что мой 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*/}
}