Swift中 Map,Flatmap,Filter,Reduce的用法

Posted 2023-04-27

参考技术A Swift的标准数组支持三个高阶函数：map,filter和reduce.是OC中没有实现的方法。

在OC中，对数组元素的操作是很不方便的一件事，遍历，操作，赋值给新数组，显得太繁琐。
在swift中，对数组的操作就简洁了很多。

map方法作用是把数组[T]通过闭包函数把每一个数组中的元素变成U类型的值，最后组成数组[U]
原型如下：
func map(transform: (T) -> U) -> [U]

在 Functional Programming in Swift 中，实现了map函数如下：

用常规方式写：

使用map：

如此简洁，尝试一下用OC实现的话。。。

flatMap 更加强大，可以传入N个处理方法，将处理后得到数据，组合到同一个数组中

filter 就是筛选的功能，参数是一个用来判断是否筛除的筛选闭包，根据闭包函数返回的Bool值来过滤值。为True则加入到结果数组中。定义如下：
func filter(includeElement: (T) -> Bool) -> [T]

常规方式实现：

使用filter实现：

一个数组
let numberArray = [1,2,3,4,5]

想要的结果
var resultArray = ["1只","2只","3只","4只","5只"]

常规方式实现：

其实swift已经比OC简洁了。想更简洁一点？用map一句话就可以了（当然，加上输入语句的话，是2句）。

reduce的作用给定一个类型为U的初始值，把数组[T]中每一个元素传入到combine的闭包函数里面，通过计算得到最终类型为U的结果值。

定义如下：
func reduce(initial: U, combine: (U, T) -> U) -> U

reduce的类似实现是：

reduce函数还可以用来实现map函数和filter函数：

常规方式实现：

使用reduce实现：

也可以写为：

如何在 RxScala/Java 中使用多个线程执行 map、filter、flatMap？

【中文标题】如何在 RxScala/Java 中使用多个线程执行 map、filter、flatMap？

【英文标题】：How to execute map, filter, flatMap using multiple threads in RxScala/Java?

【发布时间】：2016-02-18 08:35:37

【问题描述】：

如何使用多线程在Observable 上运行filter、map 和flatMap：

  def withDelay[T](delay: Duration)(t: => T) = 
    Thread.sleep(delay.toMillis)
    t
  

  Observable
    .interval(500 millisecond)
    .filter(x => 
      withDelay(1 second)  x % 2 == 0 
    )
    .map(x => 
      withDelay(1 second)  x * x 
    ).subscribe(println(_))

目标是使用多个线程同时运行过滤和转换操作。

【问题讨论】：

你看过这个：github.com/ReactiveX/RxJava/issues/1673 和这个：github.com/ReactiveX/RxJavaParallel

@david.mihola，是的，我检查了它们，并且能够在多个线程中执行 subscribe 块，但是对于 map、flatMap 和 filter，我无法做到这一点。我假设我可能会在过滤或转换时调用其他 API 或从数据库中获取其他数据，因此我想确保此代码将同时执行。

【参考方案1】：

你可以在每个操作上使用 Async.toAsync()。

它在 rxjava-async 包中

Documentation

【参考方案2】：

这将在不同的线程 (rxjava3) 中处理每个集合项。

var collect = Observable.fromIterable(Arrays.asList("A", "B", "C"))
                      .flatMap(v -> 
                         return Observable.just(v)
                                        .observeOn(Schedulers.computation())
                                        .map(v1 -> 
                                            int time = ThreadLocalRandom.current().nextInt(1000);
                                            Thread.sleep(time);
                                            return String.format("processed-%s", v1);
                                        );
                      )
                      .observeOn(Schedulers.computation())
                      .blockingStream()
                      .collect(Collectors.toList());

【参考方案3】：

您必须使用 observeOn 运算符，它将在设置运算符后定义的特定线程中执行所有下一个运算符

       /**
 * Once that you set in your pipeline the observerOn all the next steps of your pipeline will be executed in another thread.
 * Shall print
 * First step main
 * Second step RxNewThreadScheduler-2
 * Third step RxNewThreadScheduler-1
 */
@Test
public void testObservableObserverOn() throws InterruptedException 
    Subscription subscription = Observable.just(1)
            .doOnNext(number -> System.out.println("First step " + Thread.currentThread()
                    .getName()))
            .observeOn(Schedulers.newThread())
            .doOnNext(number -> System.out.println("Second step " + Thread.currentThread()
                    .getName()))
            .observeOn(Schedulers.newThread())
            .doOnNext(number -> System.out.println( "Third step " + Thread.currentThread()
                    .getName()))
            .subscribe();
    new TestSubscriber((Observer) subscription)
            .awaitTerminalEvent(100, TimeUnit.MILLISECONDS);

更多异步示例在这里https://github.com/politrons/reactive/blob/master/src/test/java/rx/observables/scheduler/ObservableAsynchronous.java

【讨论】：

是的，但是如果你使用 .doOnNext() 你不能做地图，平面等操作。它只接受动作。

这是一个例子，你应该可以使用任何运算符