响应式编程中遇上游错误不中断元素处理的实现方案

问题2：Mutiny框架场景修复 #

原代码中.onFailure().recoverWithItem(0)是针对整个流的错误恢复，流中出现第一个错误时，它会返回指定值并直接终止流，所以后续元素不会被处理。要实现单个元素错误不中断流，需要针对每个元素的转换操作单独处理错误。

推荐用onItem().handle方法，它允许手动处理每个元素的转换，遇到错误时可以选择发射默认值或忽略该元素，同时不终止整个流。

修改后的代码：

Multi.createFrom().items(1, 2, 3, 4, 5, 0, 7, 8, 9, 10)
        .onRequest().invoke(req -> System.out.println("Got a request: " + req))
        .onItem().handle((i, emitter) -> {
            try {
                // 正常计算时发射结果
                emitter.emit(100 / i);
            } catch (ArithmeticException e) {
                // 除零错误时发射0；如果要忽略该元素，改成emitter.request(1)即可
                emitter.emit(0);
            }
        })
        .subscribe().withSubscriber(new Flow.Subscriber<Integer>() {

            private Flow.Subscription subscription;

            @Override
            public void onSubscribe(Flow.Subscription s) {
                this.subscription = s;
                s.request(1);
            }

            @Override
            public void onNext(Integer item) {
                System.out.println("Got item " + item);
                subscription.request(1);
            }

            @Override
            public void onError(Throwable t) {
                System.out.println("error " + t);
            }

            @Override
            public void onComplete() {
                System.out.println("finish");
            }
        });

修改后，每个元素的转换错误被局部处理，流会继续处理后续所有元素，最终正常完成。

内容的提问来源于stack exchange，提问作者robyp7