使用vertx超文本传输协议客户端时可观察ZIP运算符挂起

Question

我所做的：

我正在使用vertx rx超文本传输协议客户端来执行大量的HTTP请求。在这种情况下，我调用“方法A”，它返回一个ID列表。要接收所有ID，我需要多次调用方法A以获取下一批结果。（每次我指定一个不同的页码我想接收）

为了提高性能并尽可能并行地进行调用，我创建了一个（RxJava）可观察项目列表，每个项目代表单个页面请求的结果。当我完成创建此列表时，我调用Obserable.zip运算符并传递可观察列表。

问题：

在没有特殊设置的情况下使用vertx http客户端，一切都可以正常工作，但速度非常慢。e、 g.在5分钟内处理3000个http请求。

我试图通过如下设置vertx http客户端选项来提高性能：

 HttpClientOptions options = new HttpClientOptions();

 options.setMaxPoolSize(50)
        .setKeepAlive(true)
        .setPipelining(true)
        .setTcpKeepAlive(true)
        .setPipeliningLimit(25)
        .setMaxWaitQueueSize(10000);

但是，当我这样做时，我会得到不稳定的结果：有时一切正常，我能够在不到20秒的时间内收到所有响应。但是，有时我都调用的外部服务器会关闭连接，日志显示以下错误：

io.vertx.core.http.impl.HttpClientRequestImpl
SEVERE: io.vertx.core.VertxException: Connection was closed

• 我的代码中没有调用错误处理程序

下面是创建HttpClientRequest的代码

public Observable<HttpRestResponse> postWithResponse(String url, Map<String, String> headers, String body) {
        Observable<HttpRestResponse> bufferObservable = Observable.create(subscriber -> {
            try {
                HttpClientRequest request = httpClient.postAbs(url);
                addHeadersToRequest(headers, request);
                sendRequest(url, subscriber, request, body);
            }catch (Exception e) {
                try {
                    subscriber.onError(e);
                }catch (Exception ex) {
                    logger.error("error calling onError for subscriber",ex);
                }finally {
                    subscriber.onCompleted();
                }
            }
        });
        return bufferObservable;
    }

private void sendRequest(String requestUrl, Subscriber<? super HttpRestResponse> subscriber, HttpClientRequest request, String bodyData) {
        final long requestId = reqNumber.getAndIncrement();

        if (bodyData != null) {
            request.putHeader("Content-Length", String.valueOf(bodyData.getBytes().length);
        }

        request.putHeader("Accept-Encoding", "gzip,deflate");

        Observable<HttpRestResponse> retVal = request.toObservable()
                .doOnError(throwable -> {
                    logger.error("<<< #: " + requestId + " HTTP call failed. requestUrl [" + requestUrl + "] reason:" + throwable.getMessage());
                }).doOnNext(response -> {
                    if (response != null) {
                        logger.debug(" <<< #: " + requestId + " " + response.statusCode() + " " + response.statusMessage() + " " + requestUrl);
                    }
                }).flatMap(httpClientResponse -> {
                    try {
                        if (httpClientResponse != null && doCheckResponse(httpClientResponse, requestUrl, requestId, bodyData)) {
                            Observable<Buffer> bufferObservable = httpClientResponse.toObservable()
                                    .reduce(Buffer.buffer(1000), (result, buffer) -> result.appendBuffer(buffer));

                            return bufferObservable.flatMap(buffer -> Observable.just(new HttpRestResponse(buffer, httpClientResponse)));
                        }
                    } catch (Exception e) {
                        logger.error("error in RestHttpClient", e);
                    }

                    return Observable.just(new HttpRestResponse(null, httpClientResponse));
                });

        retVal.subscribe(subscriber);

        if (bodyData != null) {
            request.end(bodyData); // write post data
        } else {
            request.end();
        }
    }

关于我们

Answer 1

所以最后我想通了这一点。看来我传给了Observable.zip方法一个空列表...

这里的问题是onNext或onError没有在“zip”方法返回的可观察对象上被调用。在这种情况下，只调用onComplete，我没有为它创建处理程序。。。

非常感谢所有感兴趣并试图提供帮助的人。

雅尼夫

Answer 2

如果您认为可以像这样连接异常逻辑

 try {
            HttpClientRequest request = httpClient.postAbs(url);
            addHeadersToRequest(headers, request);
            sendRequest(url, subscriber, request, body);
        }catch (Exception e) {
            try {
                subscriber.onError(e);
            }catch (Exception ex) {
                logger.error("error calling onError for subscriber",ex);
            }finally {
                subscriber.onCompleted();
            }
        }

您不会收到任何错误，因为您的所有处理现在基本上都驻留在Rx生态系统中，因此在您的try-catch块中不会向您报告任何内容。

这个时间点的错误会从你的

bufferObservable.onErrorReturn()

或

bufferObservable.subscribe(success, error)