我正在使用vert.x和RxJava在Apigateway上工作。我想发送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"
} ]
我看到您正在onNext()
中启动新线程?当您使用Thread.start()
启动新线程时,它将并行进行其工作,当前Observable.onNext()
将认为工作已完成。这将导致调用onComplete()
。
没有人等待线程的工作完成。希望你明白我的意思。
现在,我对vert.x不熟悉,但是按照RxJava的方式,似乎不需要使用Thread
。您可以删除它,并将其余的逻辑直接保留在onNext()
方法中。
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数组,所以我没有实现它。