如何使用 apache HttpClient 流式传输响应正文答案

【问题标题】：How to stream response body with apache HttpClient如何使用 apache HttpClient 流式传输响应正文
【发布时间】：2014-12-09 03:58:37
【问题描述】：

我需要一个 api 来执行八位字节流，它没有长度。它只是一个实时数据流。我遇到的问题是，当我提出请求时，它似乎试图等待内容结束，然后再将信息读入输入流，但是它没有看到内容的结束和 NoHttpResponse 异常的超时。以下是我的代码的简化版本：

private static HttpPost getPostRequest() {
    // Build uri
    URI uri = new URIBuilder()
            .setScheme("https")
            .setHost(entity.getStreamUrl())
            .setPath("/")
            .build();

    // Create http http
    HttpPost httpPost = new HttpPost(uri);

    String nvpsStr = "";
    Object myArray[] = nvps.toArray();
    for(int i = 0; i < myArray.length; i ++) {
        nvpsStr += myArray[i].toString();
        if(i < myArray.length - 1) {
            nvpsStr += "&";
        }
    }

    // Build http payload
    String request = nvpsStr + scv + streamRequest + "\n\n";
    // Attach http data
    httpPost.setEntity(new StringEntity(URLEncoder.encode(request,"UTF-8")));

    return httpPost;
}

// Where client is simply
// private static final CloseableHttpClient client = HttpClients.createDefault();
private static runPostRequest (HttpPost request) {
    CloseableHttpResponse response = client.execute(request);
    try {
        HttpEntity ent = response.getEntity();
        InputStream is = ent.getContent();
        DataInputStream dis = new DataInputStream(is);
        // Only stream the first 200 bytes
        for(int i = 0; i < 200; i++) {
            System.out.println(( (char)dis.readByte()));
        }

    } finally {
        response.close();
    }
}

【问题讨论】：

这对我来说很好用。也许向我们展示您的服务器端。
我无法控制服务器端代码。这只是我用来传输股票报价的第 3 方 API。
你能给我们举个例子吗？我尝试了一个简单的 servlet，它每隔几秒就会传输几个字节，并且接收内容没有问题。
不幸的是，我使用的 api 让我签署了 NDA。我可能会在这个问题上与他们的工作人员核实。感谢您验证此代码的有效性。
Apache HTTPClient Streaming HTTP POST Request? 的可能重复项

标签： java httpclient apache-httpclient-4.x apache-commons-httpclient

【解决方案1】：

编辑 2

所以，如果你对线程/runnables/Handlers 不满意并且对 android AsyncTask 不满意，我会直接转到 HttpUrlConnection（放弃使用 Apache HttpClient 的整个练习，因为基本上 Google 说 HttpUrlConnection 将支持流式响应它确实有效！）

检测转储标头等所有细节可能并不容易。但是对于普通的流式响应对象，我认为它应该可以正常工作....参见 HttpsUrlConnection 代码示例的编辑 3

EndEdit2

不清楚正在使用什么“流”协议（渐进式下载或 HTTP 流式传输）或您如何实际管理客户端上的流式响应。

建议从连接中转储标头以查看客户端和服务器的确切协议？？

我假设您关闭了 UI 线程（在 AsyncTask 中或在处理程序的回调部分中）；如果这不准确，您可能需要稍微重构一下。

假设 HTTP 流与 Apache HttpClient 4.3.5+ 一起使用

如果响应的标头中没有长度，那么您将在 HTTP 1.1 上执行“分块”响应，您必须读取缓冲区，直到获得“最后一个块”或决定关闭流或连接：

服务器刚刚开始发送（流式传输），客户端应按照关于生成实体内容的详细 Apache 说明使用缓冲区来处理它从 HTTP 响应中获得的“输入流”。

我不记得 30 秒的套接字超时是否会抢占活动流？请记住，在 Apache 中，构建器中存在单独的套接字超时设置和 read 超时设置。不希望套接字关闭您，也不希望在服务器提供响应时超时等待可读流的可用字节。

无论如何，客户端处理程序只需要通过检查读入缓冲区的内容来了解流如何结束...

如果现有协议是“继续”和“分块”，则客户端上的响应处理程序应该处于流处理程序循环中，直到它看到来自 http spec 的 LAST-CHUNK。

 response.getEntity().getContent()

应该给你处理响应流直到'last-chunk'所需的参考......

我认为您应该 read here 了解如何使用缓冲实体，在该实体中，需要不止一次读取才能在响应中的“最后一个块”处结束。这是 HttpURLConnection 可能更容易的另一个原因......

执行一个循环处理缓冲读取，直到匹配“last-chunk”的字节发出 END 信号。

然后按照关于消费实体和可重用连接的详细 Apache 说明关闭流或连接。

EDIT Apache HttpClient 中流式响应的代码

在'处理程序的回调或异步任务中

 request.execute();
...

 processStreamingEntity(response.getEntity());
 response.close();

//implement your own wrapper as mentioned in apache docs

    private void processStreamingEntity(HttpEntity entity) throws IOException {
        InputStreamHttpEntityHC4 bufHttpEntity = new InputStreamHttpEntityHC4(entity);
        while not bufHttpEntity.LAST_CHUNK {
            handleResponse(bufHttpEntity.readLine())
}

编辑 3

HttpURLConnection 版本，如果你这样做的话。（使用 MessageHandler 但您可以在适当的位置使用字节，因为这是来自流式语音示例，并且文本中的单词在此处被发送回 UI）

private void openHttpsConnection(String urlStr, Handler mhandler) throws IOException {
    HttpsURLConnection httpConn = null;
    String line = null;
    try {
        URL url = new URL(urlStr);
        URLConnection urlConn = url.openConnection();               
        if (!(urlConn instanceof HttpsURLConnection)) {
            throw new IOException ("URL is not an Https URL");
        }               
        httpConn = (HttpsURLConnection)urlConn;
        httpConn.setAllowUserInteraction(false);
        httpConn.setInstanceFollowRedirects(true);
        httpConn.setRequestMethod("GET");
        httpConn.setReadTimeout(50 * 1000);
        BufferedReader is =
                new BufferedReader(new InputStreamReader(httpConn.getInputStream()));                   
        
        while ((line = is.readLine( )) != null) {

                Message msg = Message.obtain();
                msg.what=1;  
                msg.obj=line;                       
                mhandler.sendMessage(msg);
            
        }               
    } catch (MalformedURLException e) {
        e.printStackTrace();
    } catch( SocketTimeoutException e){
        e.printStackTrace();
    } catch (IOException e) {

        e.printStackTrace();
        Message msg = Message.obtain();
            msg.what=2;
            BufferedInputStream in = new BufferedInputStream(httpConn.getErrorStream());
    
            line =new String(readStream(in));
            msg.obj=line;
            mhandler.sendMessage(msg);
          
    }
    finally {httpConn.disconnect();}

}

【讨论】：

它应该是 http 流媒体，直到最近我才处理过。我使用 URL.openStream() 来接收我的数据，因为它没有过早关闭我的请求......不像 httpclient 连接。
嗯。从你的代码看起来像一个 apache httpclient。 IMO 4.3.5 + 如果您将标题设置为“继续”和分块，我认为我编辑中的伪代码会消耗 response.entity。
我还没试过，我去看看。正如你所说，我只希望服务器支持 http 1.1。我将此标记为答案，因为它确实看起来像我正在寻找的东西。我会告诉你结果。谢谢！
是的。请记住，在 android 中，对于流/实体，可能仍然存在一些情况，其中 httpUrlconnection 将更好地处理包装的流。我会尝试使用 apache 来找到正确的 'entity'wrapper 。然后如果没有工作，回到httpUrlConn
注意。在非常短的流上运行带有“trace-asci”的 Curl POST 将提供一些关于细节的良好上下文。

【解决方案2】：

尝试 RxSON：https://github.com/rxson/rxson 它利用带有 RxJava 的 JsonPath 在响应到达后立即从响应中读取 JSON 流块，并在响应完成之前将它们解析为 java 对象。

例子：

String serviceURL = "https://think.cs.vt.edu/corgis/datasets/json/airlines/airlines.json";
   HttpRequest req = HttpRequest.newBuilder(URI.create(serviceURL)).GET().build();
   RxSON rxson = new RxSON.Builder().build();

   String jsonPath = "$[*].Airport.Name";
   Flowable<String> airportStream = rxson.create(String.class, req, jsonPath);
   airportStream
       .doOnNext(it -> System.out.println("Received new item: " + it))
       //Just for test
       .toList()
       .blockingGet();

【讨论】：