如何使用 Mutiny 反应式编程调用 long 运行 阻塞 void 返回方法?
How to call long running blocking void returning method with Mutiny reactive programming?
我在 Mutiny 的 Uni 上有一个 Async 和 Sync 方法调用链,一些方法是 return 类型 void
的长 运行 进程。
在不阻塞下游的情况下 invoking/calling 它们的正确方法是什么?
下面是简单的类比代码
class Root {
public static void main(String[] args) {
final Response response = getResponsePayload(); // Gets the the Payload from upstream service
Uni.createFrom().item(response)
.onItem().invoke(() -> System.out.println("Process Started"))
.onItem().call(res -> {
longRunningMethodAsync(res); // long running blocking method, I want to run on a worker thread
return Uni.createFrom().voidItem(); // This line I created, because of the ppipeline will be broken if the Uni is not returned from here
})
.onItem().transform(item -> item.hello + " mutiny")
.onItem().transform(String::toUpperCase)
.subscribe().with(
item -> System.out.println(">> " + item)); // This is printed to the console
}
// Boilerplate method - I created to invoke/call the actual method actual method - `longRunningMethod`, this method basically an adapter
// This is the best apprach I could come up, but I'm looking for better thatn this as I'm not conviced I'm doing it right
private static UniSubscribe<Void> longRunningMethodAsync(final Response response) {
return Uni.createFrom().voidItem().invoke(() -> longRunningMethod(response))
.runSubscriptionOn(Infrastructure.getDefaultExecutor()).subscribe();
}
// Important - this is the method I want to run asynchronously independently of main *event-loop* thread.
private static void longRunningMethod(final Response response) {
System.out.println("Long running process started"); // Doesn't get printed, which means this is never called at all, not even in the blocked manner by the main even-loop thread
}
// Not as importatnt, I provded this in case if you like to run on your local box
private static Response getResponsePayload() {
return new Response();
}
private static class Response {
public final String hello = "hello";
}
}
一般情况下,使用runSubscriptionOn
并传递一个特定的执行器:
longRunningMethodAsync
.runSubscriptionOn(executor);
请注意,它将并发限制为执行程序中可用的线程数。
参考:
尝试:
private static UniSubscribe<Void> longRunningMethodAsync(final Response response) {
return Uni.createFrom()
.voidItem()
.invoke(() -> longRunningMethod(response))
.runSubscriptionOn(Infrastructure.getDefaultExecutor())
.subscribeAsCompletionStage().join();
}
我在 Mutiny 的 Uni 上有一个 Async 和 Sync 方法调用链,一些方法是 return 类型 void
的长 运行 进程。
在不阻塞下游的情况下 invoking/calling 它们的正确方法是什么?
下面是简单的类比代码
class Root {
public static void main(String[] args) {
final Response response = getResponsePayload(); // Gets the the Payload from upstream service
Uni.createFrom().item(response)
.onItem().invoke(() -> System.out.println("Process Started"))
.onItem().call(res -> {
longRunningMethodAsync(res); // long running blocking method, I want to run on a worker thread
return Uni.createFrom().voidItem(); // This line I created, because of the ppipeline will be broken if the Uni is not returned from here
})
.onItem().transform(item -> item.hello + " mutiny")
.onItem().transform(String::toUpperCase)
.subscribe().with(
item -> System.out.println(">> " + item)); // This is printed to the console
}
// Boilerplate method - I created to invoke/call the actual method actual method - `longRunningMethod`, this method basically an adapter
// This is the best apprach I could come up, but I'm looking for better thatn this as I'm not conviced I'm doing it right
private static UniSubscribe<Void> longRunningMethodAsync(final Response response) {
return Uni.createFrom().voidItem().invoke(() -> longRunningMethod(response))
.runSubscriptionOn(Infrastructure.getDefaultExecutor()).subscribe();
}
// Important - this is the method I want to run asynchronously independently of main *event-loop* thread.
private static void longRunningMethod(final Response response) {
System.out.println("Long running process started"); // Doesn't get printed, which means this is never called at all, not even in the blocked manner by the main even-loop thread
}
// Not as importatnt, I provded this in case if you like to run on your local box
private static Response getResponsePayload() {
return new Response();
}
private static class Response {
public final String hello = "hello";
}
}
一般情况下,使用runSubscriptionOn
并传递一个特定的执行器:
longRunningMethodAsync
.runSubscriptionOn(executor);
请注意,它将并发限制为执行程序中可用的线程数。
参考:
尝试:
private static UniSubscribe<Void> longRunningMethodAsync(final Response response) {
return Uni.createFrom()
.voidItem()
.invoke(() -> longRunningMethod(response))
.runSubscriptionOn(Infrastructure.getDefaultExecutor())
.subscribeAsCompletionStage().join();
}