java - 使用 PipedOutputStream 和 PipedInputStream 的多线程写入器和单线程读取器

Question

我有一个具有单读者线程模型的多写入线程。ThreadMultipleDateReceiver 类设计用于从多个线程读取。

public class ThreadMultipleDateReceiver extends Thread {

  private static final int MAX_CLIENT_THREADS = 4;
  private byte[] incomingBytes;
  private volatile boolean isRunning;
  private volatile List<ThreadStreamDateWriter> lThrdDate;

  private static PipedInputStream pipedInputStream;

  public ThreadMultipleDateReceiver() {
    lThrdDate = Collections.synchronizedList(new ArrayList<>(MAX_CLIENT_THREADS));
    pipedInputStream = new PipedInputStream();
    System.out.println("ThreadMultipleDateReceiver Created");
  }

  @Override public void run() {
    isRunning = true;
    while (isRunning) {
      if (!lThrdDate.isEmpty()) {
        System.out.println("ThreadMultipleDateReceiver has:" + lThrdDate.size());
        for (int i = lThrdDate.size(); i > 0; i--) {
          if (lThrdDate.get(i - 1).getState() == Thread.State.TERMINATED) {
            lThrdDate.remove(i - 1);
          } else {
            System.out.println("I ThreadMultipleDateReceiver have:" + lThrdDate.get(i - 1).getNameDateWriter());
          }
        }
        incomingBytes = new byte[1024];
        try {
          String str = "";
          int iRd;
          System.out.println("ThreadMultipleDateReceiver waiting:" + str);
          while ((iRd = pipedInputStream.read(incomingBytes)) != -1) {
            if (iRd > 0) {
              str += new String(incomingBytes);
            }
          }
          System.out.println("ThreadMultipleDateReceiver Received:\n\t:" + str);
        } catch (IOException e) { }
      } else {
        System.out.println("ThreadMultipleDateReceiver Empty");
      }
    }
    emptyDateWriters();
  }

  public void addDateWriter(ThreadStreamDateWriter threadDateWriter) {
    if (lThrdDate.size() < MAX_CLIENT_THREADS) {
      lThrdDate.add(threadDateWriter);
    }
  }

  private void emptyDateWriters() {
    if (!lThrdDate.isEmpty()) {
      for (int i = lThrdDate.size(); i > 0; i--) {
        ThreadStreamDateWriter threadDateWriter = lThrdDate.get(i - 1);
        threadDateWriter.stopThread();
        lThrdDate.remove(i - 1);
      }
    }
  }

  public PipedInputStream getPipedInputStream() {
    return pipedInputStream;
  }

  public void stopThread() {
    isRunning = false;
  }

}

以及单个 Writer 线程

public class ThreadStreamDateWriter extends Thread {
  String Self;
  private byte[] outgoingBytes;
  private volatile boolean isRunning;
  private static PipedOutputStream pipedOutputStream;

  ThreadStreamDateWriter(String name, PipedInputStream snk) {
    Self = name;
    pipedOutputStream = new PipedOutputStream();
    try {
      pipedOutputStream.connect(snk);
    } catch (IOException e) { }
  }

  @Override public void run() {
    isRunning = true;
    while (isRunning) {
      try {
        outgoingBytes = getInfo().getBytes();
        System.out.println("ThreadStreamDateWriter -> write to pipedOutputStream:" + new String(outgoingBytes));
        pipedOutputStream.write(outgoingBytes);
        System.out.println("ThreadStreamDateWriter -> wrote:" + new String(outgoingBytes));
        try { Thread.sleep(4000); } catch (InterruptedException ex) { }
      } catch (IOException | NegativeArraySizeException | IndexOutOfBoundsException e) {
        isRunning = false;
      }
    }
  }

  String getInfo() {
      String sDtTm = new SimpleDateFormat("yyyyMMdd-hhmmss").format(Calendar.getInstance().getTime());
      return Self + " -> " + sDtTm;
  }

  public void stopThread() {
    isRunning = false;
  }

  public String getNameDateWriter() {
    return Self;
  }
}

如何启动(我使用的是 Netbeans)？

ThreadMultipleDateReceiver thrdMDateReceiver = null;
ThreadStreamDateWriter thrdSDateWriter0 = null;
ThreadStreamDateWriter thrdSDateWriter1 = null;
  private void jtbDateExchangerActionPerformed(java.awt.event.ActionEvent evt) {  
    if (jtbDateExchanger.isSelected()) {
      if (thrdMDateReceiver == null) {
        thrdMDateReceiver = new ThreadMultipleDateReceiver();
        thrdMDateReceiver.start();
      }
      if (thrdSDateWriter0 == null) {
        thrdSDateWriter0 = new ThreadStreamDateWriter("-0-", thrdMDateReceiver.getPipedInputStream());
        thrdSDateWriter0.start();
        thrdMDateReceiver.addDateWriter(thrdSDateWriter0);
      }
      if (thrdSDateWriter1 == null) {
        thrdSDateWriter1 = new ThreadStreamDateWriter("-1-", thrdMDateReceiver.getPipedInputStream());
        thrdSDateWriter1.start();
        thrdMDateReceiver.addDateWriter(thrdSDateWriter1);
      }
    } else {
      if (thrdMDateReceiver != null) {
        thrdMDateReceiver.stopThread();
      }
    }
  }                                                

输出

    run:
ThreadMultipleDateReceiver Created
ThreadMultipleDateReceiver Empty
ThreadMultipleDateReceiver Empty
ThreadMultipleDateReceiver Empty
.....
ThreadMultipleDateReceiver Empty
ThreadMultipleDateReceiver Empty
ThreadMultipleDateReceiver Empty
ThreadMultipleDateReceiver has:1
I ThreadMultipleDateReceiver have:-0-
ThreadMultipleDateReceiver waiting:
ThreadStreamDateWriter -> write to pipedOutputStream:-0- -> 20170608-090003
ThreadStreamDateWriter -> write to pipedOutputStream:-1- -> 20170608-090003
BUILD SUCCESSFUL (total time: 1 minute 3 seconds)

ThreadMultipleDateReceiver 被阻止，并且不打印:

ThreadMultipleDateReceiver Received:
    -1- -> 20170608-090003

或

ThreadMultipleDateReceiver Received:
    -0- -> 20170608-090003

如何解决？

Answer 1

看起来您的管道输出流是静态的，因此每次构造 ThreadStreamDateWriter 时，您都会使用管道输出流的旧值。

尝试将其设为实例变量并将其传递到构造函数中。所以你只有其中之一。

编辑1:我创建了管道实例变量并添加了一些打印输出。现在似乎运行时间更长(见下文):

编辑2:你第二个pipedOutputStream.connect(snk);正在 throw 。您一次只能连接一件事。

import java.io.IOException;
import java.io.PipedInputStream;
import java.io.PipedOutputStream;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Collections;
import java.util.List;
public class So44438086 {
    public static class ThreadMultipleDateReceiver extends Thread {
        private static final int MAX_CLIENT_THREADS=4;
        private byte[] incomingBytes;
        private volatile boolean isRunning;
        private volatile List<ThreadStreamDateWriter> lThrdDate;
        private /*static*/ PipedInputStream pipedInputStream;
        public ThreadMultipleDateReceiver() {
            lThrdDate=Collections.synchronizedList(new ArrayList<>(MAX_CLIENT_THREADS));
            pipedInputStream=new PipedInputStream();
            System.out.println("ctor setting pipedInputStream to: "+pipedInputStream);
            System.out.println("ThreadMultipleDateReceiver Created");
        }
        @Override public void run() {
            isRunning=true;
            while(isRunning) {
                if(!lThrdDate.isEmpty()) {
                    System.out.println("ThreadMultipleDateReceiver has:"+lThrdDate.size());
                    for(int i=lThrdDate.size();i>0;i--) {
                        if(lThrdDate.get(i-1).getState()==Thread.State.TERMINATED) {
                            lThrdDate.remove(i-1);
                        } else {
                            System.out.println("I ThreadMultipleDateReceiver have:"+lThrdDate.get(i-1).getNameDateWriter());
                        }
                    }
                    incomingBytes=new byte[1024];
                    try {
                        String str="";
                        int iRd;
                        System.out.println("ThreadMultipleDateReceiver waiting:"+str);
                        System.out.println("reading: "+pipedInputStream);
                        while((iRd=pipedInputStream.read(incomingBytes))!=-1) {
                            if(iRd>0) {
                                str+=new String(incomingBytes);
                            }
                        }
                        System.out.println("ThreadMultipleDateReceiver Received:\n\t:"+str);
                    } catch(IOException e) {}
                } else {
                    System.out.println("ThreadMultipleDateReceiver Empty");
                }
            }
            emptyDateWriters();
        }
        public void addDateWriter(ThreadStreamDateWriter threadDateWriter) {
            if(lThrdDate.size()<MAX_CLIENT_THREADS) {
                lThrdDate.add(threadDateWriter);
            }
        }
        private void emptyDateWriters() {
            if(!lThrdDate.isEmpty()) {
                for(int i=lThrdDate.size();i>0;i--) {
                    ThreadStreamDateWriter threadDateWriter=lThrdDate.get(i-1);
                    threadDateWriter.stopThread();
                    lThrdDate.remove(i-1);
                }
            }
        }
        public PipedInputStream getPipedInputStream() {
            return pipedInputStream;
        }
        public void stopThread() {
            isRunning=false;
        }
    }
    public static class ThreadStreamDateWriter extends Thread {
        String Self;
        private byte[] outgoingBytes;
        private volatile boolean isRunning;
        private /*static*/ PipedOutputStream pipedOutputStream;
        ThreadStreamDateWriter(String name,PipedInputStream snk) {
            Self=name;
            pipedOutputStream=new PipedOutputStream();
            System.out.println("ctor setting pipedOutputStream to: "+pipedOutputStream);
            try {
                pipedOutputStream.connect(snk);
                System.out.println(pipedOutputStream+" connectd to: "+snk);
            } catch(IOException e) {}
        }
        @Override public void run() {
            isRunning=true;
            while(isRunning) {
                try {
                    outgoingBytes=getInfo().getBytes();
                    System.out.println("ThreadStreamDateWriter -> write to pipedOutputStream:"+new String(outgoingBytes));
                    System.out.println("writing to: "+pipedOutputStream);
                    pipedOutputStream.write(outgoingBytes);
                    System.out.println("ThreadStreamDateWriter -> wrote:"+new String(outgoingBytes));
                    try {
                        Thread.sleep(4000);
                    } catch(InterruptedException ex) {}
                } catch(IOException|NegativeArraySizeException|IndexOutOfBoundsException e) {
                    isRunning=false;
                }
            }
        }
        String getInfo() {
            String sDtTm=new SimpleDateFormat("yyyyMMdd-hhmmss").format(Calendar.getInstance().getTime());
            return Self+" -> "+sDtTm;
        }
        public void stopThread() {
            isRunning=false;
        }
        public String getNameDateWriter() {
            return Self;
        }
    }
    private void foo() {
        if(thrdMDateReceiver==null) {
            thrdMDateReceiver=new ThreadMultipleDateReceiver();
            thrdMDateReceiver.start();
        }
        if(thrdSDateWriter0==null) {
            thrdSDateWriter0=new ThreadStreamDateWriter("-0-",thrdMDateReceiver.getPipedInputStream());
            thrdSDateWriter0.start();
            thrdMDateReceiver.addDateWriter(thrdSDateWriter0);
        }
        if(thrdSDateWriter1==null) {
            thrdSDateWriter1=new ThreadStreamDateWriter("-1-",thrdMDateReceiver.getPipedInputStream());
            thrdSDateWriter1.start();
            thrdMDateReceiver.addDateWriter(thrdSDateWriter1);
        }
    }
    void run() throws InterruptedException {
        System.out.println(("running"));
        foo();
        System.out.println(("sleeping"));
        Thread.sleep(10000);
        System.out.println(("stopping"));
        if(thrdMDateReceiver!=null) {
            thrdMDateReceiver.stopThread();
        }
    }
    public static void main(String[] args) throws InterruptedException {
        new So44438086().run();
    }
    ThreadMultipleDateReceiver thrdMDateReceiver=null;
    ThreadStreamDateWriter thrdSDateWriter0=null;
    ThreadStreamDateWriter thrdSDateWriter1=null;
}