Я столкнулся с очень странной проблемой.
Я работаю над Vert.x, и из обработчика я вызываю API REST, используя HttpClientRequest из Vert.x.Теперь у меня есть CompletableFuture, который я заполняю в обработчике ответа HttpClientRequest.Позже я использую CompletableFuture.get().Но всякий раз, когда вызывается метод get(), основной поток блокируется (как и ожидалось), но он остается заблокированным навсегда.Я не вижу обратного вызова в моем обработчике ответов, и он застрял навсегда.
Вот код:
import io.vertx.core.http.HttpClientRequest;
import io.vertx.core.http.HttpMethod;
import io.vertx.core.json.Json;
import io.vertx.core.json.JsonObject;
import java.util.concurrent.CompletableFuture;
import io.vertx.core.http.HttpClient;
CompletableFuture<JsonObject> comp = new CompletableFuture<JsonObject>();
HttpClient httpClient = new HttpClient(); //This object initialized and set the endpoit, port and domain name.
HttpClientRequest request = httpClient.request(HttpMethod.POST, requestURI, response -> {
response.bodyHandler(body -> {
//do some process
comp.complete(new JsonObject(body);
});
}).exceptionHandler(e -> {
//log the error
comp.completeExceptionally(e);
});
request.end();
//after some process
comp.get(); // here main thread is stuck forever.
Мой API дает ответ 200, я видел в нем Wireshark, а также, если я делаю comp.thenAccept(), выполняется обратный вызов, и он дает мой результат.
Почему это происходит и каково решение?
Примечание: я знаю, что не рекомендуется использовать метод Completable.get(), но в моем случае я должен его использовать.
Вот пример кода, который вызывает у меня проблему:
package io.vertx.starter;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.Future;
import io.vertx.core.http.*;
import io.vertx.ext.web.Router;
import io.vertx.ext.web.RoutingContext;
import java.util.concurrent.CompletableFuture;
public class SampleVerticle extends AbstractVerticle {
public void start ( Future startFuture ) throws Exception {
Future<Void> future = Future.future ();
HttpServer server = vertx.createHttpServer ();
Router router = Router.router (vertx);
router.get ("/sample").handler (this::sampeHandler);
router.get ("/testcompletableblocking").handler (this::testCompBlocking);
router.get ("/testcompletablenonblocking").handler (this::testCompNonBlocking);
server.requestHandler (router::accept) // <5>
.listen (8080, ar -> { // <6>
if (ar.succeeded ()) {
System.out.println ("Server started");
future.complete ();
} else {
System.out.println ("Server is not started");
future.fail (ar.cause ());
}
});
}
private void sampeHandler ( RoutingContext context ) {
try {
Thread.sleep (1000);
} catch (Exception e) {
}
String response = "Hello...";
context.response ().setStatusCode (200).putHeader ("content-type", "text/html").end (response);
}
private void testCompBlocking ( RoutingContext context ) {
System.out.println ("Calling testCompBlocking....");
HttpClientOptions clientOptions = new HttpClientOptions ().setDefaultHost ("localhost").setDefaultPort (8080).setSsl (false).setKeepAlive (true);
HttpClient client = vertx.createHttpClient (clientOptions);
String requestURI = "/sample";
CompletableFuture<String> comp = new CompletableFuture<> ();
HttpClientRequest request = client.request (HttpMethod.GET, requestURI, response -> {
response.bodyHandler (body -> {
String kmsResponse = new String (body.getBytes ());
System.out.println ("kmsResponse-" + kmsResponse);
comp.complete (kmsResponse);
});
}).exceptionHandler (e -> {
e.printStackTrace ();
comp.completeExceptionally (e);
});
request.end ();
String result = "Not Success";
try {
result = comp.get ();
} catch (Exception e) {
System.out.println ("Exception in getting from Completable..." + e.getMessage ());
e.printStackTrace ();
}
context.response ().setStatusCode (200);
context.response ().putHeader ("content-type", "text/html");
context.response ().end (result);
System.out.println ("end testCompBlocking....");
}
private void testCompNonBlocking ( RoutingContext context ) {
System.out.println ("Calling testCompNonBlocking....");
HttpClientOptions clientOptions = new HttpClientOptions ().setDefaultHost ("localhost").setDefaultPort (8080).setKeepAlive (false);
HttpClient client = vertx.createHttpClient (clientOptions);
String requestURI = "/sample";
CompletableFuture<String> comp = new CompletableFuture<> ();
HttpClientRequest request = client.request (HttpMethod.GET, requestURI, response -> {
response.bodyHandler (body -> {
String kmsResponse = new String (body.getBytes ());
System.out.println ("kmsResponse-" + kmsResponse);
comp.complete (kmsResponse);
});
}).exceptionHandler (e -> {
e.printStackTrace ();
comp.completeExceptionally (e);
});
request.end ();
String result = "Not Blocking, please see result at Console";
try {
comp.thenAccept (apiResult -> System.out.println ("apiResult from CompletableFuture - " + apiResult));
} catch (Exception e) {
System.out.println ("Exception in getting from Completable..." + e.getMessage ());
e.printStackTrace ();
}
context.response ().setStatusCode (200);
context.response ().putHeader ("content-type", "text/html");
context.response ().end (result);
System.out.println ("end testCompNonBlocking....");
}
}
Вызов localhost:8080/testcompletableblocking, ответ не отправлен, и текущий поток заблокирован навсегда.