[Day-24] - Spring Reactor Mono 一日初探就上手

Abstract

昨日已先行提过Flux，可方便处理一连串指定类型事件，所以说Flux就像瑞凡，被众人讨厌的瑞凡，散播出去，就难收回了，所以我们说，瑞凡，回不去了，而Ｍono用於处理单一指定类型事件，可说是一种单一(Single)架构，其中也包含了许多与Flux相似的静态方法，如：just、empty、error及never等用法，所以只要出错位置都可以确切地发现，而Ｍono有许多针对自身元件设计的特别方法，较常被开发者使用的有fromCallable、fromCompletionStage、fromFuture、fromRunnable和fromSupplier方法，这五种方法分别从Callable、CompletionStage、CompletableFuture、Runnable和Supplier中创建了Mono对象，我是小编威斯丁，我将以fromCallable作原理与范例介绍。

Principle Introduction

Mono架构是处理0到1个异步发射器，他是采用一种非阻塞式的架构，可以确保每个指定类型事件可以运算完毕，不受其他执行绪的影响，Mono是一个专一的Publisher，也就是说最多只能发射出一个类别，可在运作元subscribe任务後，以onComplete方法或onError方法结束。可用於操作Flux子集，可将Ｍono与另一个发布者的操作与做组合，并切换到Flux资料序列，如Mono.concatWith(Publisher)进行返回一个Flux ，而Mono.then(Mono)则返回一个Mono，亦可用於表示只有执行概念且无返回物件值地执行绪。如：Mono，我们可参照以下流程图，可得知粗黑垂直线为onComplete方法，红色Ｘ代表onError方法，仅支援一次性发射与运作性任务，提供给各位开发者作参考。

图一、Mono 物件运作流程图

小编延续先前范例，将台湾区及中国区建立产品的程序码区段，改为Mono单一执行绪处理方式，确版建立过程不受其他程序影响，可看出在成功时(onComplete)会写出建立成功的log，错误时(onError)会喷出错误例外及写错误Log。

  // Taiwan create product flow
    @Override
    public SeaFood createSeaFood(SeaFood seaFood) throws SeaFoodRetailerGenericException {
       validateNullId(seaFood);
        Mono.fromCallable(() ->seaFood).subscribe(
                new Consumer<SeaFood>() {
                    @Override
                    public void accept(SeaFood seaFood1) {
                        SEA_FOOD_CACHE_TAIWAN.asMap().putIfAbsent(seaFood.getId(),seaFood);
                    }
                },
                new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) {
                        try {
                            logger.error("Taiwan product create fail. ex:{}",throwable.getMessage());
                            throw new SeaFoodRetailerGenericException("Out of EXPECT error.");
                        } catch (SeaFoodRetailerGenericException e) {
                            e.printStackTrace();
                        }
                    }
                },
                ()->logger.info("Create Taiwan product success ! "));
        return seaFood;
    }
  
  //china create product flow 
  @Override
    public SeaFood createSeaFood(SeaFood seaFood) throws SeaFoodRetailerGenericException {
        validateNullId(seaFood);
        validateNullId(seaFood);
        Mono.fromCallable(() ->seaFood).subscribe(
                new Consumer<SeaFood>() {
                    @Override
                    public void accept(SeaFood seaFood1) {
                        SEA_FOOD_CACHE_CHINESE.asMap().putIfAbsent(seaFood.getId(),seaFood);
                    }
                },
                new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) {
                        try {
                            logger.error("China product create fail. ex:{}",throwable.getMessage());
                            throw new SeaFoodRetailerGenericException("Out of EXPECT error.");
                        } catch (SeaFoodRetailerGenericException e) {
                            e.printStackTrace();
                        }
                    }
                },
                ()->logger.info("Create China product success ! "));

        return seaFood;
    }

透过Postman发射Create all Product测试结果，可看到以下成功的log，可看出我们一个session发射出两个执行绪(Thread)。

23:14:08.476  INFO 21055 --- [nio-8080-exec-2] o.s.web.servlet.DispatcherServlet        : Completed initialization in 10 ms
23:14:08.651  INFO 21055 --- [nio-8080-exec-2] s.s.s.s.SeaFoodRetailerServiceImpl       : Create Taiwan product success ! 
23:14:08.654  INFO 21055 --- [nio-8080-exec-2] s.s.s.s.ChinaSeaFoodRetailerServiceImpl  : Create China product success ! 
Subscribe model : {"id":"C-0099","name":"A-cha Crab","description":"Opilio is the primary species referred to as A-cha crab."}

小编透过以上流程与说明提供给各位开发者作参考，架构各位可参考昨日所带出的Flux资料流架构，都是共通继承CoreSubscriber介面，仅差别在Mono而外拉出专用的LambdaMonoSubscriber介面，并再客制其特有的方法，其所有处理的静态方法都是透过onAssembly进行转换後回传新Mono对象，为一种方法链(Method Chain)写法。