Reactor操作符的共享与复用

在 Reactor 中，transform 和 transformDeferred 是两个用于代码复用和操作符链封装的高级操作符。它们允许你将一组操作符封装成一个函数，并在适当的时候应用到响应式流中。以下是它们的详细总结：

1. transform 操作符

• 作用：transform 操作符允许你将操作符链的一部分封装成一个函数，并在实例化时（assembly time）应用到原始操作符链上。这意味着所有订阅者都会应用相同的操作符链。
• 适用场景：适用于所有订阅者共享相同行为的场景，例如统一的日志记录、转换等。
• 示例：

    // 定义一个函数，用于过滤掉"orange"颜色并将其余颜色转换为大写
    Function<Flux<String>, Flux<String>> filterAndMap =
            f -> f.filter(color -> !color.equals("orange"))
                    .map(String::toUpperCase);
    
    // 从列表创建Flux，并在过滤和映射前打印每个元素
    Flux.fromIterable(Arrays.asList("blue", "green", "orange", "purple"))
            .doOnNext(System.out::println)
            // 使用之前定义的函数对Flux进行转换
            .transform(filterAndMap)
            // 订阅转换后的Flux，并打印每个元素
            .subscribe(d -> System.out.println("Subscriber to Transformed MapAndFilter: " + d));
  

  执行结果如下

  blue
  Subscriber to Transformed MapAndFilter: BLUE
  green
  Subscriber to Transformed MapAndFilter: GREEN
  orange
  purple
  Subscriber to Transformed MapAndFilter: PURPLE
  

2. transformDeferred 操作符

• 作用：transformDeferred 操作符与 transform 类似，也允许你将操作符链封装成一个函数，但主要的区别在于函数的执行时机：transformDeferred 是在订阅发生时（subscription time）才应用函数的。这意味着它可以为每个订阅者生成不同的操作链，甚至可以维护一些状态（例如计数器、随机数等）。
• 适用场景：适用于每个订阅者需要不同操作链的场景，例如根据订阅者的身份、时间、状态等动态生成不同的操作链。
• 示例：

    // 使用AtomicInteger来实现线程安全的计数器
    AtomicInteger ai = new AtomicInteger();
    
    // 定义一个函数，根据计数器的值来选择不同的处理逻辑
    Function<Flux<String>, Flux<String>> filterAndMap = f -> {
        // 如果计数器的值为1，则过滤掉"orange"颜色，否则过滤掉"purple"颜色
        if (ai.incrementAndGet() == 1) {
            return f.filter(color -> !color.equals("orange"))
                    .map(String::toUpperCase);
        }
        return f.filter(color -> !color.equals("purple"))
                .map(String::toUpperCase);
    };
    
    // 创建一个Flux，从颜色列表开始，通过.doOnNext()打印每个元素，然后通过.transformDeferred()应用filterAndMap函数
    Flux<String> composedFlux =
            Flux.fromIterable(Arrays.asList("blue", "green", "orange", "purple"))
                    .doOnNext(System.out::println)
                    .transformDeferred(filterAndMap);
    
    // 订阅第一个处理流程的结果
    composedFlux.subscribe(d -> System.out.println("Subscriber 1 to Composed MapAndFilter :" + d));
    // 订阅第二个处理流程的结果
    composedFlux.subscribe(d -> System.out.println("Subscriber 2 to Composed MapAndFilter: " + d));
  

  执行结果如下

  blue
  Subscriber 1 to Composed MapAndFilter :BLUE
  green
  Subscriber 1 to Composed MapAndFilter :GREEN
  orange
  purple
  Subscriber 1 to Composed MapAndFilter :PURPLE
  blue
  Subscriber 2 to Composed MapAndFilter: BLUE
  green
  Subscriber 2 to Composed MapAndFilter: GREEN
  orange
  Subscriber 2 to Composed MapAndFilter: ORANGE
  purple
  

3. transform 与 transformDeferred 的区别

4. 总结

• transform：适用于所有订阅者共享相同操作链的场景，例如统一的日志记录、转换等。
• transformDeferred：适用于每个订阅者需要不同操作链的场景，例如根据订阅者的身份、时间、状态等动态生成不同的操作链。

通过使用 transform 和 transformDeferred，你可以更灵活地测试响应式流，尤其是当你需要模拟复杂或非标准的数据流时，它们是非常有用的工具。