6

Apache HTTPClientを使用して「全二重」HTTP ストリーミング リクエストを構築しようとしています。

最初の試みでは、次のリクエスト コードを使用してみました。

URL url=new URL(/* code goes here */);

HttpPost request=new HttpPost(url.toString());

request.addHeader("Connection", "close");

PipedOutputStream requestOutput=new PipedOutputStream();
PipedInputStream requestInput=new PipedInputStream(requestOutput, DEFAULT_PIPE_SIZE);
ContentType requestContentType=getContentType();
InputStreamEntity requestEntity=new InputStreamEntity(requestInput, -1, requestContentType);
request.setEntity(requestEntity);

HttpEntity responseEntity=null;
HttpResponse response=getHttpClient().execute(request); // <-- Hanging here
try {
    if(response.getStatusLine().getStatusCode() != 200)
        throw new IOException("Unexpected status code: "+response.getStatusLine().getStatusCode());

    responseEntity = response.getEntity();
}
finally {
    if(responseEntity == null)
        request.abort();
}

InputStream responseInput=responseEntity.getContent();
ContentType responseContentType;
if(responseEntity.getContentType() != null)
    responseContentType = ContentType.parse(responseEntity.getContentType().getValue());
else
    responseContentType = DEFAULT_CONTENT_TYPE;

Reader responseStream=decode(responseInput, responseContentType);
Writer requestStream=encode(requestOutput, getContentType());

要求は、上記の行でハングします。コードは、応答を取得する前に要求全体を送信しようとしているようです。振り返ってみると、これは理にかなっています。しかし、それは私が望んでいたものではありません。:)

代わりに、 でリクエスト ヘッダーを送信し、独自のヘッダーでのTransfer-Encoding: chunkedレスポンス ヘッダーを受信して​​から、全二重ストリーミング HTTP 接続を使用することを望んでいました。HTTP/1.1 200 OKTransfer-Encoding: chunked

幸いなことに、私の HTTPClient には別の NIO ベースの非同期クライアントがあり、適切な使用例 (このようなもの) があります。私の質問は次のとおりです。

  1. 同期 HTTPClient の動作に対する私の解釈は正しいですか? または、説明した方法で (より単純な) 同期 HTTPClient を引き続き使用するためにできることはありますか?
  2. NIO ベースのクライアントは、応答を求める前に要求全体を送信するのを待ちますか? または、リクエストを段階的に送信し、同時に応答を段階的に受信することはできますか?

HTTPClient がこのモダリティをサポートしない場合、別の HTTP クライアント ライブラリはありますか? または、このモダリティをサポートするために (最小限の) HTTP クライアントを作成することを計画する必要がありますか?

4

2 に答える 2

0

コードが機能するには、別のスレッドが OutputStream に書き込んでいる必要があります。

  • 上記のコードは、HTTPClient に PipedInputStream を提供します。
  • PipedInputStream は、バイトが対応する OutputStream に書き込まれるときに使用できるようにします。
  • 上記のコードは、OutputStream に書き込みません (別のスレッドで行う必要があります。
  • したがって、コードはコメントがある場所に正確にぶら下がっています。
  • 内部では、Apache クライアントは "inputStream.read()" を言います。これは、パイプされたストリームの場合、outputStream.write(bytes) が以前に (別のスレッドによって) 呼び出されている必要があります。
  • 別のスレッドから関連する OutputStream にバイトを送り込むわけではないため、InputStream はただ座って、「他のスレッド」によって OutputStream に書き込まれるのを待ちます。

JavaDocs から:

パイプされた入力ストリームは、パイプされた出力ストリームに接続する必要があります。パイプされた入力ストリームは、パイプされた出力ストリームに書き込まれたデータ バイトを提供します。

通常、データは 1 つのスレッドによって PipedInputStream オブジェクトから読み取られ、データは別のスレッドによって対応する PipedOutputStream に書き込まれます。

スレッドがデッドロックする可能性があるため、1 つのスレッドから両方のオブジェクトを使用しようとすることはお勧めしません。

パイプされた入力ストリームにはバッファが含まれており、制限内で読み取り操作を書き込み操作から分離しています。接続されたパイプ出力ストリームにデータ バイトを提供していたスレッドが生きていない場合、パイプは「壊れている」と言われます。

注:パイプストリームと同時実行性は問題のステートメントで言及されていないため、必要ではないように思えます。最初に健全性チェックのために、代わりに Entity オブジェクトで ByteArrayInputStream() をラップしてみてください...これは、問題を絞り込むのに役立ちます。

アップデート

ちなみに、 Apache Commons HTTP Client 4.3.4 を使用して、HTTP POST 用の OutputStream インターフェイスを提供するApache の HTTP クライアント API [PipedApacheClientOutputStream]の反転を作成しました。これはあなたが探しているものに近いかもしれません...

呼び出しコードは次のようになります。

// Calling-code manages thread-pool
ExecutorService es = Executors.newCachedThreadPool(
  new ThreadFactoryBuilder()
  .setNameFormat("apache-client-executor-thread-%d")
  .build());


// Build configuration
PipedApacheClientOutputStreamConfig config = new      
  PipedApacheClientOutputStreamConfig();
config.setUrl("http://localhost:3000");
config.setPipeBufferSizeBytes(1024);
config.setThreadPool(es);
config.setHttpClient(HttpClientBuilder.create().build());

// Instantiate OutputStream
PipedApacheClientOutputStream os = new     
PipedApacheClientOutputStream(config);

// Write to OutputStream
os.write(...);

try {
  os.close();
} catch (IOException e) {
  logger.error(e.getLocalizedMessage(), e);
}

// Do stuff with HTTP response
...

// Close the HTTP response
os.getResponse().close();

// Finally, shut down thread pool
// This must occur after retrieving response (after is) if interested   
// in POST result
es.shutdown();

注-実際には、同じクライアント、エグゼキュータ サービス、および構成がアプリケーションの存続期間を通じて再利用される可能性が高いため、上記の例の外側の準備および終了コードは、直接インライン化されるのではなく、ブートストラップ/init およびファイナライズ コードに存在する可能性があります。 OutputStream のインスタンス化。

于 2016-01-27T00:48:27.720 に答える