Vertx/RxJava/WebClient/ApiGateway/Reactive

Question

我正在使用vert.x和RxJava开发Apigate。我想发送2个Api的响应请求，从它们两个获得响应，并通过HttpServer发送组合的JSON。但是onComplete()执行到早期并返回空的JSON。我认为问题来自vert.x的asynchronous字符，但我不知道到底出了什么问题。

这是我的方法：

private void dispatchBoth(RoutingContext routingContext) {
Observer<String> observer = new Observer<String>() {
JsonArray jsonArray = new JsonArray();
@Override
public void onSubscribe(Disposable disposable) {
System.out.println("Start");
}
@Override
public void onNext(String s) {
Thread t = new Thread(() -> {
if(s=="/api/userApi/selectAllUsers") {
WebClient client = WebClient.create(vertx);
client
.get(8081, "localhost", s)
.send(ar->{
if (ar.succeeded()) {
HttpResponse<Buffer> response = ar.result();
jsonArray.addAll(response.bodyAsJsonArray());
System.out.println(jsonArray.encodePrettily());
} else {
System.out.println("Something went wrong " + ar.cause().getMessage());
}
});
}else if(s=="/api/holidayApi/selectAllHolidays") {
WebClient client = WebClient.create(vertx);
client
.get(8080, "localhost", s)
.send(ar -> {
if (ar.succeeded()) {
HttpResponse<Buffer> response = ar.result();
jsonArray.addAll(response.bodyAsJsonArray());
//  System.out.println(jsonArray.encodePrettily());
} else {
System.out.println("Something went wrong " + ar.cause().getMessage());
}
});
}
});
t.start();
}
@Override
public void onError(Throwable throwable) {
}
@Override
public void onComplete() {
System.out.println(jsonArray.encodePrettily());
routingContext.response().end(jsonArray.encodePrettily());
}
};
Observable.fromArray(com).subscribe(observer);
}

这就是我在控制台上得到的输出：

[ ]
[ {
"holidayId" : 2,
"userId" : 3,
"place" : "Poland",
"date" : {
"year" : 2016,
"month" : "DECEMBER",
"dayOfMonth" : 29,
"dayOfWeek" : "THURSDAY",
"era" : "CE",
"dayOfYear" : 364,
"leapYear" : true,
"monthValue" : 12,
"chronology" : {
"id" : "ISO",
"calendarType" : "iso8601"
}
}
}, {
"holidayId" : 10,
"userId" : 1,
"place" : "Netherland",
"date" : {
"year" : 2020,
"month" : "JANUARY",
"dayOfMonth" : 21,
"dayOfWeek" : "TUESDAY",
"era" : "CE",
"dayOfYear" : 21,
"leapYear" : true,
"monthValue" : 1,
"chronology" : {
"id" : "ISO",
"calendarType" : "iso8601"
}
}
}, {
"userId" : 1,
"name" : "Kacper",
"phone_number" : "667667202"
}, {
"userId" : 3,
"name" : "Kamil",
"phone_number" : "6734583443"
}, {
"userId" : 4,
"name" : "Janek",
"phone_number" : "231253575"
}, {
"userId" : 5,
"name" : "Grzegorz",
"phone_number" : "123456789"
}, {
"userId" : 6,
"name" : "Justin",
"phone_number" : "111000111"
}, {
"userId" : 8,
"name" : "Mike",
"phone_number" : "997"
}, {
"userId" : 9,
"name" : "Gorge",
"phone_number" : "991"
} ]

Answer 1

onComplete按时执行：当字符串的输入流结束时。您需要另一种方式来等待所有I/O操作完成。这是一个棘手的部分，我不知道Vertx或RxJava是否可以做到这一点，但标准Java API可以使用CompletableFuture。因此，我们创建了从CompletableFuture到Handler的适配器，它保存一个I/O操作的结果：

class HandelerFuture extends CompletableFuture<JsonArray> 
implements Handler<AsyncResult<HttpResponse<Buffer>>> {
@Override
public void handle(AsyncResult<HttpResponse<Buffer>> ar) {
if (ar.succeeded()) {
JsonArray array = ar.result().bodyAsJsonArray();
super.complete(array);
} else {
super.completeExceptionally(ar.cause());
}
}
}

此外，您不需要将onNext方法的主体包裹在Tread中。其次，您不需要检查传递了什么url字符串，因为它没有任何区别。第三，您只使用Observer来处理url字符串列表——这太复杂了。简单的循环就足够了。

CompletableFuture<HandelerFuture[]> dispatchBoth(String... urls) {
ArrayList<HandelerFuture> futures = new ArrayList<>(); // all results
for (String url : urls) {
HandelerFuture future = new HandelerFuture();
futures.add(future);
WebClient client = WebClient.create(vertx);
client.get(8081, "localhost", url)
.send(future);
}
CompletableFuture all = new CompletableFuture();
HandelerFuture[] array = futures.toArray(new HandelerFuture[0]);
CompletableFuture.allOf(array)
.thenRunAsync(() -> all.complete(array));
return all;
}

则可以按以下方式运行：

CompletableFuture<HandelerFuture[]> future = dispatchBoth(com);
HandelerFuture[] results = future.get();
JsonArray finalArray;
for (HandelerFuture result:results) {
try {
// extract partial json array
JsonArray partialArray = result.get();
// combine partialArray with finalArray somehow
} catch (Exception e) {
// this is the exception got in handle() method as ar.cause().
e.printStackTrace();
}
}
routingContext.response().end(finalArray.encodePrettily());

您没有告诉我们将如何组合json数组，所以我没有实现它。

Answer 2

我看到您正在onNext()中启动一个新线程？当您使用Thread.start()启动一个新线程时，它将并行地进行工作，而当前的Observable.onNext()将认为工作已完成。这将导致调用onComplete()。

没有人在等待线程的作业完成。希望你明白我的意思。

现在，我不熟悉vert.x，但按照RxJava的方式，似乎没有必要使用Thread。您可以删除它，并将逻辑的其余部分直接保留在onNext()方法中。