如何从 HttpsURLConnection 创建 Java 非阻塞 InputStream？

Question

基本上，我有一个 URL，可以在发布新消息时从聊天室流式传输 xml 更新。我想将该 URL 转换为 InputStream 并继续从中读取，只要保持连接并且只要我没有发送 Thread.interrupt()。我遇到的问题是当有内容要从流中读取时，BufferedReader.ready() 似乎不会成为真的。

我正在使用以下代码：

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

对于非阻塞 IO，不要使用 InputStream 和 Reader（或 OutputStream/Writer），而是使用 java.nio.* 类，在本例中为 SocketChannel（以及额外的 CharsetDecoder）。

---

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

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

套接字（以及 SocketChannel）在传输层 (TCP) 上工作，比 HTTP 等应用层协议低一层（或两层）。所以你不能创建一个到 https url 的套接字通道。

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

这个is quite complicated（我做过一次），如果你没有实现一个同时连接很多客户端的服务器（你不能有一个线程），你真的不想这样做每个连接）。相反，保留从 URLConnection 读取的阻塞 InputStream，并将其简单地放在不会妨碍应用程序其余部分的备用线程中。

Answer 2

如何创建Java非阻塞InputStream

你不能。您的问题在术语上体现了矛盾。 Java 中的流是阻塞的。因此不存在“非阻塞InputStream”之类的东西。

Reader.ready() 在可以无阻塞地读取数据时返回 true。时期。 InputStreams 和 Readers 正在阻塞。时期。这里的一切都按设计工作。如果您希望这些 API 具有更高的并发性，您将不得不使用多个线程。或Socket.setSoTimeout() 及其在HttpURLConnection 中的近亲。

Answer 3

您可以使用提供非阻塞 I/O 功能的 Java NIO 库。详情和示例代码请看这篇文章：http://www.drdobbs.com/java/184406242。

Answer 4

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

}

Answer 5

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