将错误从 PipedWriter+PipedReader 对的写入线程委托给读取线程

Question

我想要什么？

我有一个带有两个线程的 PipedWriter + PipedReader 对。我希望，在写入器线程中捕获的异常将在包装到 IOException 中的 PipedReader 的下一个操作中抛出。如果 PipedReader 本身有错误，则可以将写入线程异常添加到被抑制的异常中。

为什么？

问题是，如果写入线程中发生错误，我只能关闭写入器，这会导致读取器端出现正常的 EOF。我希望读者线程了解作者方面的错误。但我希望我的代码的用户应该看到一个简单的阅读器。所以我必须将该错误作为 IOException 呈现。

问题

你知道有什么开箱即用的“错误委派 pipedwriter+reader”实现吗？你将如何以线程安全的方式实现这样的事情？

Answer 1

解决方案

我创建了一个“ErrorDelegatingReaderDecorator”：

import java.io.IOException;
import java.io.Reader;

class ErrorDelegatingReaderDecorator extends Reader {
    static final String ERROR_MESSAGE_WRITER_THREAD = "Error occoured on the other side of the pipe. See the cause!";
    static final String ERROR_MESSAGE_READER_THREAD = "Error occoured on the this side of the pipe. See the cause!";
    private Reader decorated;
    private Throwable delegatedThrowable;

    public ErrorDelegatingReaderDecorator(Reader decorated) {
        super();
        this.decorated = decorated;
    }

    @Override
    public int read(char[] cbuf, int off, int len) throws IOException {
        Throwable originalException = null;
        int result = -1;
        try {
             result = decorated.read(cbuf, off, len);           
        }
        catch( Throwable original ) {
            originalException = original;
        }
        finally {
            throwException(originalException);
        }
        return result;
    }

    @Override
    public void close() throws IOException {
        Throwable originalException = null;
        try {
            decorated.close();          
        }
        catch( Throwable original ) {
            originalException = original;
        }
        finally {
            throwException(originalException);
        }
    }

    private synchronized void throwException(Throwable originalException) throws IOException {
        Throwable delegatedTemp = delegatedThrowable;
        delegatedThrowable = null;
        if ( originalException != null ) {
            if ( delegatedTemp != null ) {
                originalException.addSuppressed(delegatedTemp);
            }
            throw new IOException( ERROR_MESSAGE_READER_THREAD, originalException ) ;
        }
        else if ( delegatedTemp != null ) {
            throw new IOException( ERROR_MESSAGE_WRITER_THREAD, delegatedTemp );
        }
    }


    public synchronized void setDelegatedThrowable(Throwable delegatedThrowable) {
        this.delegatedThrowable = delegatedThrowable;
    }
}

然后我可以这样使用它：

    final PipedWriter pipedWriter = new PipedWriter();
    PipedReader pipedReader = new PipedReader( pipedWriter, pipeBufferSize);
    final ErrorDelegatingReaderDecorator errorDelegatingReader = new ErrorDelegatingReaderDecorator( pipedReader );

    executorService.execute( new Runnable( ) {
        @Override
        public void run() {
            try 
            {
                //do something that writes into the writer and can throw an error
            } catch (Exception e) {
                errorDelegatingReader.setDelegatedThrowable(e);
                LOGGER.error( "Error while processing excel file.", e );
            }
            finally {
                try {
                    pipedWriter.close();
                } catch (IOException e) {
                    LOGGER.error( "Error while closing printwriter.", e );
                }
            }
        }
    });
    //Giving out the decorated reader to the reader thread.

权衡

由于我必须从修饰阅读器的方法的 finally 部分抛出异常，因此我必须捕获 Throwable，并且必须将其包装到 IOException 中。它有点难看，但它不太可能发生在经过装饰的PipedReader 上。

为什么它没有先工作

正如我还告诉我的有趣的事情一样，我是如何在返回 EOF 而不是抛出错误时遇到间歇性失败的：我在 writer 块中使用了 try-with-resources。结果是我的作家首先被关闭，然后我在装饰器中设置了错误。如果在两者之间我有一个阅读，那么它就变成了一个EOF。所以我用普通的 finally 块替换了它，现在顺序没问题。