如何从 HttpsURLConnection 创建 Java 非阻塞输入流？

Question

基本上，我有一个 URL，用于在发布新消息时从聊天室传输 xml 更新。我想将该 URL 转换为 InputStream，并在保持连接且未发送 Thread.interrupt() 的情况下继续从中读取数据。我遇到的问题是，当有内容要从流中读取时，BufferedReader.ready() 似乎不会变为 true。

我使用以下代码：

BufferedReader buf = new BufferedReader(new InputStreamReader(ins));


String str = "";
while(Thread.interrupted() != true)
{
    connected = true;
    debug("Listening...");

    if(buf.ready())
    {
        debug("Something to be read.");
        if ((str = buf.readLine()) != null) {
            // str is one line of text; readLine() strips the newline character(s)
            urlContents += String.format("%s%n", str);
            urlContents = filter(urlContents);
        }
    }

    // Give the system a chance to buffer or interrupt.
    try{Thread.sleep(1000);} catch(Exception ee) {debug("Caught thread exception.");}
}

当我运行代码并将某些内容发布到聊天室时， buf.ready() 永远不会变为 true，导致这些行永远不会被读取。但是，如果我跳过“buf.ready()”部分并直接读取行，它将阻止进一步的操作，直到读取行。

我如何 a) 让 buf.ready() 返回 true，或 b) 以防止阻塞的方式执行此操作？

提前致谢， 詹姆斯

Answer 1

如何创建Java非阻塞InputStream

你不能。你的问题体现了一个术语上的矛盾。 Java 中的流是阻塞的。因此，不存在“非阻塞InputStream”这样的东西。

当可以无阻塞地读取数据时，Reader.ready() 返回 true。时期。 InputStreams 和 Readers 正在阻塞。时期。这里的一切都按设计进行。如果您希望这些 API 具有更多并发性，则必须使用多个线程。或者 Socket.setSoTimeout() 及其在 HttpURLConnection 中的密切关系。

Answer 2

对于非阻塞 IO，不要使用 InputStream 和 Reader（或 OutputStream/Writer），而是使用 java.nio.* 类，在本例中是 SocketChannel（以及附加的字符集解码器）。

---

编辑：作为对您评论的答复：

专门寻找如何创建到 https url 的套接字通道。

套接字（以及 SocketChannel）在传输层 (TCP) 上工作，传输层位于应用层协议（例如 HTTP）之下的一（或两）层。因此您无法创建到 https url 的套接字通道。

相反，您必须打开到正确服务器和正确端口的 Socket-Channel（如果 URI 中没有给出其他内容，则为 443），在客户端模式下创建 SSLEngine（在 javax.net.ssl 中），然后从通道读取数据，将其提供给 SSL 引擎，反之亦然，并向您的 SSLEngine 发送/获取正确的 HTTP 协议行，始终检查返回值以了解实际上处理了多少字节以及下一步是什么拿。

这相当复杂（我做到了一次），如果您没有实现同时连接大量客户端的服务器（其中每个连接不能有一个线程），那么您实际上并不想这样做。相反，保留从 URLConnection 读取的阻塞 InputStream，并将其简单地放在不会妨碍应用程序其余部分的备用线程中。

Answer 3

您可以使用提供非阻塞 I/O 功能的 Java NIO 库。请参阅本文以获取详细信息和示例代码： http://www.drdobbs.com/java/184406242 。

Answer 4

没有使用通道的 HTTP/HTTPS 实现。无法以非阻塞方式从 httpurlconnaction 读取输入流。您要么必须使用第三方库，要么自己通过 SocketChannel 实现 http。

Answer 5

import java.io.InputStream;
import java.util.Arrays;

/**
 * This code demonstrates non blocking read from standard input using separate
 * thread for reading.
 */
public class NonBlockingRead {

    // Holder for temporary store of read(InputStream is) value
    private static String threadValue = "";

    public static void main(String[] args) throws InterruptedException {

        NonBlockingRead test = new NonBlockingRead();

        while (true) {
            String tmp = test.read(System.in, 100);
            if (tmp.length() > 0)
                System.out.println(tmp);
            Thread.sleep(1000);
        }
    }

    /**
     * Non blocking read from input stream using controlled thread
     * 
     * @param is
     *            — InputStream to read
     * @param timeout
     *            — timeout, should not be less that 10
     * @return
     */
    String read(final InputStream is, int timeout) {

        // Start reading bytes from stream in separate thread
        Thread thread = new Thread() {

            public void run() {
                byte[] buffer = new byte[1024]; // read buffer
                byte[] readBytes = new byte[0]; // holder of actually read bytes
                try {
                    Thread.sleep(5);
                    // Read available bytes from stream
                    int size = is.read(buffer);
                    if (size > 0)
                        readBytes = Arrays.copyOf(buffer, size);
                    // and save read value in static variable
                    setValue(new String(readBytes, "UTF-8"));
                } catch (Exception e) {
                    System.err.println("Error reading input stream\nStack trace:\n" + e.getStackTrace());
                }
            }
        };
        thread.start(); // Start thread
        try {
            thread.join(timeout); // and join it with specified timeout
        } catch (InterruptedException e) {
            System.err.println("Data were note read in " + timeout + " ms");
        }
        return getValue();

    }

    private synchronized void setValue(String value) {
        threadValue = value;
    }

    private synchronized String getValue() {
        String tmp = new String(threadValue);
        setValue("");
        return tmp;
    }

}