Обработка Исключения в потоках трубопровода java

Question

У меня есть xmlStream, который я конвертирую в jsonStream, используя org.apache.wink.json4j.utils.XML. Вот код

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(); 
    } 
}