调用多个平级服务,按照服务优先级返回第一个有效数据。
具体case:一个页面可能有很多的弹窗,弹窗之间又有优先级。每次只需要返回第一个有数据的弹窗。但是又希望所有弹窗之间的数据获取是异步的。这种场景使用 Reactor 怎么实现呢?
public interface TestServiceI { Mono request(); }
提供一个 request 方法,返回一个 Mono 对象。
@Data @ToString @AllArgsConstructor @NoArgsConstructor public class TestUser { private String name; }
@Slf4j public class TestServiceImpl1 implements TestServiceI { @Override public Mono request() { log.info("execute.test.service1"); return Mono.fromSupplier(() -> { try { System.out.println("service1.threadName=" + Thread.currentThread().getName()); Thread.sleep(500); } catch (InterruptedException e) { throw new RuntimeException(e); } return ""; }) .map(name -> { return new TestUser(name); }); } }
第一个 service 执行耗时 500ms。返回空对象;
创建第二个 service 执行耗时 1000ms。返回空对象;代码如上,改一下sleep时间即可。
继续创建第三个 service 执行耗时 1000ms。返回 name3。代码如上,改一下 sleep 时间,以及返回为 name3。
public static void main(String[] args) { long startTime = System.currentTimeMillis(); TestServiceI testServiceImpl4 = new TestServiceImpl4(); TestServiceI testServiceImpl5 = new TestServiceImpl5(); TestServiceI testServiceImpl6 = new TestServiceImpl6(); List<TestServiceI> serviceIList = new ArrayList<>(); serviceIList.add(testServiceImpl4); serviceIList.add(testServiceImpl5); serviceIList.add(testServiceImpl6); // 执行 service 列表,这样有多少个 service 都可以 Flux<Mono<TestUser>> monoFlux = Flux.fromIterable(serviceIList) .map(service -> { return service.request(); }); // flatMap(或者flatMapSequential) + map 实现异常继续下一个执行 Flux flux = monoFlux.flatMapSequential(mono -> { return mono.map(user -> { TestUser testUser = JsonUtil.parseJson(JsonUtil.toJson(user), TestUser.class); if (Objects.nonNull(testUser) && StringUtils.isNotBlank(testUser.getName())) { return testUser; } // null 在 reactor 中是异常数据。 return null; }) .onErrorContinue((err, i) -> { log.info("onErrorContinue={}", i); }); }); Mono mono = flux.elementAt(0, Mono.just("")); Object block = mono.block(); System.out.println(block + "blockFirst 执行耗时ms:" + (System.currentTimeMillis() - startTime)); }
执行输出:
20:54:26.512 [main] DEBUG reactor.util.Loggers - Using Slf4j logging framework
20:54:26.553 [main] INFO com.geniu.reactor.TestServiceImpl1 - execute.test.service1
service1.threadName=main
20:54:27.237 [main] INFO com.geniu.reactor.TestReactorOrderV2 - onErrorContinue=TestUser(name=)
20:54:27.237 [main] INFO com.geniu.reactor.TestServiceImpl2 - execute.test.service2
service5.threadName=main
20:54:28.246 [main] INFO com.geniu.reactor.TestReactorOrderV2 - onErrorContinue=TestUser(name=)
20:54:28.246 [main] INFO com.geniu.reactor.TestServiceImpl3 - execute.test.service3
service6.threadName=main
TestUser(name=name3)blockFirst 执行耗时ms:2895
总结:这样实现按照顺序返回第一个正常数据。但是执行并没有异步。下一步:如何实现异步呢?
修改 service 实现。增加 .subscribeOn(Schedulers.boundedElastic())
如下:
@Slf4j public class TestServiceImpl1 implements TestServiceI { @Override public Mono request() { log.info("execute.test.service1"); return Mono.fromSupplier(() -> { try { System.out.println("service1.threadName=" + Thread.currentThread().getName()); Thread.sleep(500); } catch (InterruptedException e) { throw new RuntimeException(e); } return ""; }) //增加subscribeOn .subscribeOn(Schedulers.boundedElastic()) .map(name -> { return new TestUser(name); }); } }
再次执行输出如下:
21:02:04.213 [main] DEBUG reactor.util.Loggers - Using Slf4j logging framework
21:02:04.265 [main] INFO com.geniu.reactor.TestServiceImpl1 - execute.test.service1
service4.threadName=boundedElastic-1
21:02:04.300 [main] INFO com.geniu.reactor.TestServiceImpl2 - execute.test.service2
21:02:04.302 [main] INFO com.geniu.reactor.TestServiceImpl3 - execute.test.service3
service2.threadName=boundedElastic-2
service3.threadName=boundedElastic-3
21:02:04.987 [boundedElastic-1] INFO com.geniu.reactor.TestReactorOrderV2 - onErrorContinue=TestUser(name=)
21:02:05.307 [boundedElastic-2] INFO com.geniu.reactor.TestReactorOrderV2 - onErrorContinue=TestUser(name=)
TestUser(name=name6)blockFirst 执行耗时ms:1242
boundedElastic
;修改 service 实现,使用 CompletableFuture 执行耗时操作(这里是sleep,具体到项目中可能是外部接口调用,DB 操作等);然后使用 Mono.fromFuture 返回 Mono 对象。
@Slf4j public class TestServiceImpl1 implements TestServiceI{ @Override public Mono request() { log.info("execute.test.service1"); CompletableFuture<String> uCompletableFuture = CompletableFuture.supplyAsync(() -> { try { System.out.println("service1.threadName=" + Thread.currentThread().getName()); Thread.sleep(500); } catch (InterruptedException e) { throw new RuntimeException(e); } return "testname1"; }); return Mono.fromFuture(uCompletableFuture).map(name -> { return new TestUser(name); }); } }
执行返回如下:
21:09:59.465 [main] DEBUG reactor.util.Loggers - Using Slf4j logging framework
21:09:59.510 [main] INFO com.geniu.reactor.TestServiceImpl2 - execute.test.service2
service2.threadName=ForkJoinPool.commonPool-worker-1
21:09:59.526 [main] INFO com.geniu.reactor.TestServiceImpl3 - execute.test.service3
service3.threadName=ForkJoinPool.commonPool-worker-2
21:09:59.526 [main] INFO com.geniu.reactor.TestServiceImpl1 - execute.test.service1
service1.threadName=ForkJoinPool.commonPool-worker-3
21:10:00.526 [ForkJoinPool.commonPool-worker-1] INFO com.geniu.reactor.TestReactorOrder - onErrorContinue=TestUser(name=)
21:10:00.538 [ForkJoinPool.commonPool-worker-2] INFO com.geniu.reactor.TestReactorOrder - onErrorContinue=TestUser(name=)
TestUser(name=testname1)blockFirst 执行耗时ms:1238