Example analysis of pipeline communication in Java multi-thread programming

This article mainly introduces the communication between threads in java multi-thread programming in detail, and discusses the use of pipelines for communication. It has certain reference value. Interested friends can refer to

One chapter talked about wait/notify communication. In this section we will discuss the use of pipes for communication.

Java provides an IO stream that makes it easy for us to operate data. PipeStream is a special stream used to directly transmit data between different threads. One thread sends data to the output pipe and another thread reads data from the input pipe. Communication through pipes does not require the use of temporary files.

Java provides four classes to enable communication between threads:

①Byte stream: PipeInputStream, PipedOutputStream
②Character stream: PipedReader, PipedWriter

Let’s take a look at the implementation of byte stream:

package pipeInputOutput;
//输出流
import java.io.IOException;
import java.io.PipedOutputStream;
public class WriteDate {
 public void writeMethod(PipedOutputStream out) {
  try {
   System.out.println("write:");
   for(int i=0;i<300;i++) {
    String outDate=""+(i+1);
    out.write(outDate.getBytes());
    System.out.print(outDate);
   }
   System.out.println();
   out.close();
  }catch(IOException e) {
   e.printStackTrace();
  }
 }
}

package pipeInputOutput;
//输入流
import java.io.IOException;
import java.io.PipedInputStream;

public class ReadDate {
 public void ReadDate(PipedInputStream input) {
  try {
   System.out.println("read:");
   byte[] byteArray=new byte[20];
   int readLength=input.read(byteArray);
   while(readLength!=-1) {
    String newDate=new String(byteArray,0,readLength);
    System.out.print(newDate);
    readLength=input.read(byteArray);
   }
   System.out.println();
   input.close();
  }catch(IOException e){
   e.printStackTrace();
  }
 }
}

package pipeInputOutput;
import java.io.PipedOutputStream;
//输出线程
public class ThreadWrite extends Thread {
 private WriteDate write;
 private PipedOutputStream out;

 public ThreadWrite(WriteDate write,PipedOutputStream out) {
  super();
  this.write=write;
  this.out=out;
 }
 public void run() {
  write.writeMethod(out);
 }

}

package pipeInputOutput;
import java.io.PipedInputStream;
//输入线程
public class ThreadRead extends Thread{
 private ReadDate read;
 private PipedInputStream in;
 public ThreadRead(ReadDate read,PipedInputStream in) {
  super();
  this.read=read;
  this.in=in;
 }
 public void run() {
  read.ReadDate(in);
 }

}

package pipeInputOutput;
import java.io.IOException;
import java.io.PipedInputStream;
import java.io.PipedOutputStream;
//测试方法
public class Run {
 public static void main(String[] args) {
  try {
   WriteDate write=new WriteDate();
   ReadDate read=new ReadDate();
   PipedInputStream inputStream=new PipedInputStream();
   PipedOutputStream outputStream=new PipedOutputStream();
   //输出流与输入流进行连接。
   outputStream.connect(inputStream);
   //inputStream.connect(outputStream);
   ThreadRead readThread=new ThreadRead(read,inputStream);
   readThread.start();//先启动输出线程
   Thread.sleep(2000);
   ThreadWrite writeThread=new ThreadWrite(write,outputStream);
   writeThread.start();//后启动输入线程
  } catch (IOException e) {
   e.printStackTrace();
  } catch (InterruptedException e) {
   e.printStackTrace();
  }
 }

}

Console output:

read:
write:
123456789101112131415161718192021...
123456789101112131415161718192021...

In the above test, the input thread is started first, and then because no thread is written, the thread is blocked until data is written.

Let’s continue to look at the implementation of character stream:

package pipeInputOutput1;
import java.io.IOException;
import java.io.PipedWriter;
//字符输出流
public class WriteDate {
 public void writeMethod(PipedWriter out) {
  try {
   System.out.println("write:");
   for(int i=0;i<300;i++) {
    String outDate=""+(i+1);
    out.write(outDate);
    System.out.print(outDate);
   }
   System.out.println();
   out.close();
  }catch(IOException e) {
   e.printStackTrace();

  }
 }

}

package pipeInputOutput1;
import java.io.IOException;
import java.io.PipedReader;
//字符输入流
public class ReadDate {
 public void readMethod(PipedReader in) {

  try {
   System.out.println("read:");
   char[] byteArray=new char[20];
   int readLength=in.read(byteArray);
   while(readLength!=-1) {
    String newDate=new String(byteArray,0,readLength);
    System.out.print(newDate);
    readLength=in.read(byteArray);
   }
   System.out.println();
   in.close();
  } catch (IOException e) {
   e.printStackTrace();
  }
 }

}

package pipeInputOutput1;
import java.io.PipedWriter;
//输出流线程
public class WriteThread extends Thread {
 private WriteDate write;
 private PipedWriter out;
 public WriteThread(WriteDate write,PipedWriter out) {
  super();
  this.write=write;
  this.out=out;
 }

 public void run() {
  write.writeMethod(out);
 }

}

package pipeInputOutput1;
import java.io.PipedReader;
//输入流线程
public class ReadThread extends Thread{
 private ReadDate read;
 private PipedReader in;
 public ReadThread(ReadDate read,PipedReader in) {
  super();
  this.read=read;
  this.in=in;
 }
 public void run() {
  read.readMethod(in);
 }

}

package pipeInputOutput1;
import java.io.IOException;
import java.io.PipedReader;
import java.io.PipedWriter;
//测试方法
public class run {
 public static void main(String[] args) {
  try {
   WriteDate write=new WriteDate();
   ReadDate read=new ReadDate();

   PipedWriter out=new PipedWriter();
   PipedReader in=new PipedReader();
   //连接输出流与输入流
   out.connect(in);
   //in.connect(out);
   ReadThread threadread=new ReadThread(read,in);
   threadread.start();

   Thread.sleep(2000);
   WriteThread threadwrite=new WriteThread(write,out);
   threadwrite.start();
  } catch (IOException e) {
   e.printStackTrace();
  } catch (InterruptedException e) {
   e.printStackTrace();
  }
 }
}

Character stream and byte stream are similar. In the above example, the character stream does not need to create a byte array.

The above is the detailed content of Example analysis of pipeline communication in Java multi-thread programming. For more information, please follow other related articles on the PHP Chinese website!