1 場(chǎng)景

調(diào)用多個(gè)平級(jí)服務(wù)，按照服務(wù)優(yōu)先級(jí)返回第一個(gè)有效數(shù)據(jù)。

具體case：一個(gè)頁(yè)面可能有很多的彈窗，彈窗之間又有優(yōu)先級(jí)。每次只需要返回第一個(gè)有數(shù)據(jù)的彈窗。但是又希望所有彈窗之間的數(shù)據(jù)獲取是異步的。這種場(chǎng)景使用 Reactor 怎么實(shí)現(xiàn)呢？

2 創(chuàng)建 service

2.1 創(chuàng)建基本接口和實(shí)體類(lèi)

public interface TestServiceI {
    Mono request();
}

提供一個(gè) request 方法，返回一個(gè) Mono 對(duì)象。

@Data
@ToString
@AllArgsConstructor
@NoArgsConstructor
public class TestUser {
    private String name;
}

2.2 創(chuàng)建 service 實(shí)現(xiàn)

@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);
                });
    }
}

第一個(gè) service 執(zhí)行耗時(shí) 500ms。返回空對(duì)象；

創(chuàng)建第二個(gè) service 執(zhí)行耗時(shí) 1000ms。返回空對(duì)象；代碼如上，改一下sleep時(shí)間即可。

繼續(xù)創(chuàng)建第三個(gè) service 執(zhí)行耗時(shí) 1000ms。返回 name3。代碼如上，改一下 sleep 時(shí)間，以及返回為 name3。

3 主體方法

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);

    // 執(zhí)行 service 列表，這樣有多少個(gè) service 都可以
        Flux<Mono<TestUser>> monoFlux = Flux.fromIterable(serviceIList)
                .map(service -> {
                    return service.request();
                });
    // flatMap(或者flatMapSequential) + map 實(shí)現(xiàn)異常繼續(xù)下一個(gè)執(zhí)行
        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 中是異常數(shù)據(jù)。
                        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 執(zhí)行耗時(shí)ms：" + (System.currentTimeMillis() - startTime));
    }

• 1、Flux.fromIterable 執(zhí)行 service 列表，可以隨意增刪 service 服務(wù)。
• 2、flatMap(或者flatMapSequential) + map + onErrorContinue 實(shí)現(xiàn)異常繼續(xù)下一個(gè)執(zhí)行。具體參考：Reactor中的onErrorContinue 和 onErrorResume
• 3、Mono mono = flux.elementAt(0, Mono.just("")); 返回第一個(gè)正常數(shù)據(jù)。

執(zhí)行輸出：

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 執(zhí)行耗時(shí)ms：2895

• 1、service1 和 service2 因?yàn)榉祷乜?，所以繼續(xù)下一個(gè)，最終返回 name3。
• 2、查看總耗時(shí)：2895ms。service1 耗時(shí) 500，service2 耗時(shí)1000，service3 耗時(shí) 1000。發(fā)現(xiàn)耗時(shí)基本上等于 service1 + service2 + service3 。這是怎么回事呢？查看返回執(zhí)行的線(xiàn)程，都是 main。

總結(jié)：這樣實(shí)現(xiàn)按照順序返回第一個(gè)正常數(shù)據(jù)。但是執(zhí)行并沒(méi)有異步。下一步：如何實(shí)現(xiàn)異步呢？

4 實(shí)現(xiàn)異步

4.1 subcribeOn 實(shí)現(xiàn)異步

修改 service 實(shí)現(xiàn)。增加 .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);
                });
    }
}

再次執(zhí)行輸出如下：

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 執(zhí)行耗時(shí)ms：1242

• 1、發(fā)現(xiàn)具體實(shí)現(xiàn) sleep 的線(xiàn)程都不是 main 線(xiàn)程，而是 boundedElastic；
• 2、最終執(zhí)行耗時(shí) 1242ms，只比執(zhí)行時(shí)間最長(zhǎng)的 service2 和 service3 耗時(shí) 1000ms，多一些。證明是異步了。

4.2 CompletableFuture 實(shí)現(xiàn)異步

修改 service 實(shí)現(xiàn)，使用 CompletableFuture 執(zhí)行耗時(shí)操作（這里是sleep，具體到項(xiàng)目中可能是外部接口調(diào)用，DB 操作等）；然后使用 Mono.fromFuture 返回 Mono 對(duì)象。

@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);
        });
    }
}

執(zhí)行返回如下：

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 執(zhí)行耗時(shí)ms：1238

• 1、耗時(shí)操作都是使用 ForkJoinPool 線(xiàn)程池中的線(xiàn)程執(zhí)行。
• 2、最終耗時(shí)和方法1基本差不多。

到此這篇關(guān)于Reactor 多任務(wù)并發(fā)執(zhí)行且結(jié)果按順序返回第一個(gè)的文章就介紹到這了,更多相關(guān)Reactor 多任務(wù)執(zhí)行內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論