java - 处理 java 管道流中的异常

Question

我有一个 xmlStream，我正在使用 org.apache.wink.json4j.utils.XML 将其转换为 jsonStream。这是代码

public void process(InputStream xmlStream) {
    final BufferedInputStream bufferedXmlStream = new BufferedInputStream(xmlStream);

    PipedInputStream pipedJsonInputStream = new PipedInputStream();
    final PipedOutputStream jsonStream = new PipedOutputStream(pipedJsonInputStream);
    Thread xmlToJsonThread = new Thread(new Runnable() {
        @Override
        public void run() {
            // put your code that writes data to the outputstream here.
            try {
                XML.toJson(bufferedXmlStream, jsonStream, true);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    });
    xmlToJsonThread.setDaemon(true);
    xmlToJsonThread.start();

    //process data from piped stream
    BufferedReader reader = new BufferedReader(new InputStreamReader(
            pipedJsonInputStream, StandardCharsets.UTF_8));
    try {
        // use reader to further process json in main thread...
        parseJsonStream(reader);
    } finally {
        reader.close();
        jsonStream.close();
    }
}

当 XML.toJson 抛出异常时，我看到主线程没有退出。我该如何处理这个问题？你们认为这是将 XML 流转换为 Json 流以进行进一步处理的好方法吗？我真的很感激任何建议。多谢!

Answer 1

问题是子线程上的异常需要报告回主线程。除非您将读取器和写入器放在单独的线程中，否则这可能会很棘手。

我这样做的方法是将这两部分包装在 ExecutionCompletionService 中 - 像这样:

public void process(InputStream xmlStream) {
    ExecutorService threadPool = Executors.newFixedThreadPool(2);
    ExecutorCompletionService<Void> ecs = new ExecutorCompletionService<>(threadPool);

    final BufferedInputStream bufferedXmlStream = new BufferedInputStream(xmlStream);

    PipedInputStream pipedJsonInputStream = new PipedInputStream();
    final PipedOutputStream jsonStream = new PipedOutputStream(pipedJsonInputStream);

    ecs.submit( new Callable<Void>() {
       @Override
       public Void call() {
          // put your code that writes data to the outputstream here.
          try {
              XML.toJson(bufferedXmlStream, jsonStream, true);
          } catch (Exception e) {
              e.printStackTrace();
              throw e;
          }
          return null;
        }
    });

    ecs.submit( new Callable<Void>() {
       @Override
       public Void call() {
          try {
              // use reader to further process json in main thread...
              parseJsonStream(reader);
          } finally {
              reader.close();
              jsonStream.close();
          }
          return null;
      }
    });

    // Wait for all tasks to be done.
    // Kill the other thread if one hits an exception.
    try {
        for (int i = 0; i < 2; ++i) {
            ecs.take().get();
        }
    } finally {
        threadPool.shutdownNow();
    } 
}