问题描述
我在 Mutiny 的 Uni 上有一个 Async 和 Sync 方法调用链，有些方法是一个长期运行的过程，返回类型为 void。
下面是简单的类比代码。
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);
请注意，它会将并发限制为 executor 中可用的线程数。
参考：