java中的线程安全InputStream

Thread safe InputStream in java

提问by Manas Pratim Chamuah

How to create a Thread safe InputStream .During multithreaded operations the inputStream data gets corrupted,so how can i make my inputStream thread safe.will the following code work

如何创建线程安全的 InputStream 。在多线程操作期间 inputStream 数据被损坏，那么我如何使我的 inputStream 线程安全。以下代码是否有效

public class SynchronizedInputStream  extends InputStream{

    private InputStream in;

    private SynchronizedInputStream( InputStream in ) {
        this.in = in;
    }

    /* ... method for every InputStream type to use */
    public  static InputStream createInputStream( InputStream in) {
        return new SynchronizedInputStream( in);
    }

    public static InputStream createPushBackInputStream(InputStream in,int BUFSIZE){
        return new SynchronizedInputStream(new PushbackInputStream(in,BUFSIZE));
    }

    /* Wrap all InputStream methods Used */

    public int read(){
        synchronized (this) {
            try {
                return in.read();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return 0;
    }

    @Override
    public int available() {
        synchronized( this ) {
            try {
                return in.available();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return 0;
    }


}

In the NANOHTTPD file

在 NANOHTTPD 文件中

 public HTTPSession(TempFileManager tempFileManager, InputStream inputStream, OutputStream outputStream, InetAddress inetAddress) {
            this.inputStream=(PushbackInputStream) SynchronizedInputStream.createPushBackInputStream(inputStream);
        /*lines of code..........*/
        }

Then i call it like this

然后我这样称呼它

String Data = readStream(session.getInputStream());//session is HTTPSession
/*.....code....*/
private String readStream(InputStream in) {
        synchronized (in) {
            PushbackInputStream inputStream = (PushbackInputStream) in;
            StringBuffer outputBuffer = null;
            try {
                                 //Reading the InputStream Here
                }
            } catch (IOException ioe) {
                //error msg
            }
                return outputBuffer.toString();

        } 
    }

采纳答案by Stephen C

The short answer is that the class you have shown us is thread-safe, but the code that uses your class probably won't be thread safe!!

简短的回答是您向我们展示的类是线程安全的，但是使用您的类的代码可能不是线程安全的！！

What you have implemented are operations that atomically read one character, and atomically test if there is something to be read. The implementation of those operations is thread-safe, if (and only if) all threads use the same SynchronizedInputStreamobject to access a given InputStream, and nothing apart from your wrapper access the InputStreamdirectly.

您实现的是原子地读取一个字符的操作，并原子地测试是否有要读取的内容。这些操作的实现是线程安全的，如果（且仅当）所有线程使用相同的SynchronizedInputStream对象来访问给定的InputStream，并且除了您的包装器之外没有任何东西InputStream直接访问。

However, this most likely that this will not be sufficient to make your application's use of the streams thread-safe in the larger sense.

但是，这很可能不足以使您的应用程序在更大的意义上使用线程安全的流。

I expect that the "corruption" that you are observing is actually happening a higher level; e.g. two threads that are simultaneously making a sequence of readcalls to read (say) messages are interleaving so that some bytes of a message are going to the wrong thread. Assuming that that is your problem, then this does not fix it. Your readmethod only locks the stream while a thread reads a single byte. After unlocking, there is nothing to stop a different thread from reading the next byte.

我希望你所观察到的“腐败”实际上发生在更高的层次上；例如，同时进行一系列read调用以读取（例如）消息的两个线程是交错的，因此消息的某些字节将进入错误的线程。假设这是您的问题，那么这并不能解决问题。您的read方法仅在线程读取单个字节时锁定流。解锁后，没有什么可以阻止不同的线程读取下一个字节。

There are a few ways to solve this. For example"

有几种方法可以解决这个问题。例如”

• A simple way is to restructure your code only one thread everreads from a given InputStream. That thread reads the messages, and turns them into objects that can be handed off to others via a queue ... for example.

• Another way is to replace your wrapper class with one that reads an entire message atomically. Don't extend InputStream. Instead design your API in terms of the larger scale operations, and synchronize at that level of granularity.

• 一个简单的方法，就是要调整你的代码只有一个线程不断从给定读取InputStream。该线程读取消息，并将它们转换为可以通过队列传递给其他人的对象......例如。

• 另一种方法是将您的包装器类替换为原子地读取整个消息的包装器类。不要扩展InputStream。而是根据更大规模的操作来设计您的 API，并在该粒度级别进行同步。

UPDATE

更新

Re the extra code you added.

重新添加您添加的额外代码。

It looks likeonly one thread (the current request thread) should ever be reading from the input stream. If you are only using one thread there should beno issues with multi-threading or thread safety. (And besides, that this the way that the nanoHTTPD code is designed to work.)

它看起来像只有一个线程（当前请求的线程）不应当从输入流中读取。如果您只使用一个线程，那么多线程或线程安全应该没有问题。（此外，这是 nanoHTTPD 代码的工作方式。）

Supposing that there were multiple threads, your synchronized (in) {block in readStreamwould normallybe sufficient to make the code thread-safe, provided that all all of the threads were using the same inobject.

假设有多个线程，如果所有线程都使用相同的对象，则您的synchronized (in) {块readStream通常足以使代码线程安全in。

The problem is that your hacked HttpSessionclass is creating a separate SynchronizedInputStreamfor each "session", and THAT is what your code synchronizes on. So if (somehow) two threads created HttpSessionsobjects using the same socket input stream, they would synchronize on different objects, and there would be no mutual exclusion.

问题是您被黑的HttpSession课程SynchronizedInputStream为每个“会话”创建了一个单独的课程，这就是您的代码同步的内容。因此，如果（以某种方式）两个线程HttpSessions使用相同的套接字输入流创建对象，它们将在不同的对象上同步，并且不会相互排斥。

But this is all conjecture. So far, you have not demonstrated that there are multiple threads attempting to use the same input stream.

但这都是猜想。到目前为止，您还没有证明有多个线程尝试使用相同的输入流。

回答by Enno Shioji

You need to think about how it would make sense. Imagine more than one people are reading a magic book, which erases the character the first time anyone sees it. So only one person can read any given character. That's kind of how streams are.

你需要考虑它如何有意义。想象一下，不止一个人在读一本魔法书，当任何人第一次看到它时，它就会抹去这个角色。所以只有一个人可以阅读任何给定的字符。流就是这样。

This makes it really hard to read the book in a useful manner. When most naively done, each person will just get some random subset of the characters; not very useful information.

这使得以有用的方式阅读本书变得非常困难。当最天真地完成时，每个人只会得到一些随机的字符子集；不是很有用的信息。

One straight forward solution is to let one read it and then copy it onto a book that doesn't erase characters when one reads it. This way everyone can read the book. In some situation, you don't need everyone to understand the book, and people can just work as long as they are given a sentence. In this case, the one reader can post each sentence to a queue from which everyone takes one sentence at a time.

一个直接的解决方案是让一个人阅读它，然后将其复制到一本阅读时不会删除字符的书上。这样大家就可以看书了。在某些情况下，你不需要每个人都理解这本书，人们只要给他们一句话就可以工作。在这种情况下，一个读者可以将每个句子发布到一个队列中，每个人一次从中获取一个句子。

Other approaches include having a buffer where each threads store the character they read, and then check each time if they can form a word, and if so emitting the word for downstream processing. For an example, see Netty's codec package (e.g. this).

其他方法包括让每个线程存储它们读取的字符的缓冲区，然后每次检查它们是否可以形成一个单词，如果可以，则发出该单词以供下游处理。例如，请参阅 Netty 的编解码器包（例如this）。

These approaches are however usually implemented on top of a stream rather than inside it. You could well have a stream that does these inside, but it will probably confuse people.

然而，这些方法通常是在流的顶部而不是在流内部实现的。你很可能有一个在里面做这些的流，但它可能会让人们感到困惑。