(参数) -> 表达式

(参数) ->{ 代码语句 }

List<String> names = Arrays.asList("aaa", "cccc", "ddd", "bbb");
Collections.sort(names, new Comparator<String>() {
    @Override
    public int compare(String a, String b) {
        return b.compareTo(a);
    }
});

Collections.sort(names, (String a, String b) -> {
    return b.compareTo(a);
});
// 只有一条逻辑语句，可以省略大括号
Collections.sort(names, (String a, String b) -> b.compareTo(a));
// 可以省略入参类型
Collections.sort(names, (a, b) -> b.compareTo(a));

Stream.of("我是字符串A", "我是字符串B").map(s -> s.length()).filter(l -> l == 3);

// 静态方法引用
Stream.of(someStringArray).allMatch(StringUtils::isNotEmpty);
// 实例对象上的方法引用
Stream.of(someStringArray).map(this::someTransform);
// 类上的方法引用
Stream.of(someStringArray).mapToInt(String::length);
// 构造方法引用
Stream.of(someStringArray).collect(Collectors.toCollection(LinkedList::new));
// 数组构造方法引用
Stream.of(someStringArray).toArray(String[]::new);

        Function<T, R> stringIntegerFunction //输入T返回R的函数
        Predicate<T> predicate //输入T，返回boolean值，断言（谓词）函数
        Consumer<T> consumer; //消费者函数，消费一个数据
        Supplier<T> supplier; // 生产者函数，提供数据

    /**
     * 将范型T对象应用到输入的参数上，然后返回计算结果
     *
     * @param t the function argument
     * @return the function result
     */
    R apply(T t);

   /**
     *  返回一个先执行before函数对象apply方法再执行当前函数对象apply方法的函数对象
     *
     * @param <V> 前置函数的的输入类型，以及函数的输入类型 由函数
     *
     */
    default <V> Function<V, R> compose(Function<? super V, ? extends T> before) {
        Objects.requireNonNull(before);
        return (V v) -> apply(before.apply(v));
    }


  /**
    * 返回一个先执行当前函数对象apply方法再执行after函数对象apply方法的函数对象。
    *  <br>
    *    compose 和 andThen 的不同之处是函数执行的顺序不同。compose 函数先执行参数，
    *     然后执行调用者，而 andThen 先执行调用者，然后再执行参数。
    *  </br>
    */
   default <V> Function<T, V> andThen(Function<? super R, ? extends V> after) {
        Objects.requireNonNull(after);
        return (T t) -> after.apply(apply(t));
    }

   /**
     *  返回输入结果
     */
    static <T> Function<T, T> identity() {
        return t -> t;
    }

/**
 * 发布者（生产者）接口。
 * <p>
 * 用于定义一个可以发布数据流的组件。发布者可以被订阅多次，每次订阅都会生成一个新的订阅者（Subscriber）。
 * 发布者负责生成数据，并通过订阅者的方法（如 {@link Subscriber#onNext(Object)}）将数据推送给订阅者。
 * <p>
 * 如果在执行过程中出现错误，发布者会调用订阅者的 {@link Subscriber#onError(Throwable)} 方法，
 * 并且不会继续发送数据。
 * <p>
 * 示例：
 * <pre>
 * Publisher<String> publisher = subscriber -> {
 *     subscriber.onNext("Hello");
 *     subscriber.onNext("World");
 *     subscriber.onComplete();
 * };
 * </pre>
 *
 * @param <T> 发布的数据类型
 */
public interface Publisher<T> {
    /**
     * 订阅方法。
     * <p>
     * 该方法用于将一个订阅者（Subscriber）绑定到当前发布者（Publisher）。
     * 当订阅者调用此方法时，发布者会开始生成数据，并通过订阅者的方法（如 {@link Subscriber#onNext(Object)}）
     * 将数据推送给订阅者。
     * <p>
     * 注意：每个订阅者只能订阅一次发布者，且订阅后发布者会为每个订阅者生成独立的数据流。
     *
     * @param s 订阅者
     */
    public void subscribe(Subscriber<? super T> s);
}



/**
 * 订阅者（消费者）接口。
 * <p>
 * 用于定义一个可以接收数据流的组件。订阅者通过订阅发布者（Publisher）来接收数据。
 * 订阅者需要实现以下方法来处理数据流：
 * <ul>
 *     <li>{@link #onSubscribe(Subscription)}：在订阅发布者后执行，用于接收订阅关系。</li>
 *     <li>{@link #onNext(Object)}：接收发布者发送的下一个数据。</li>
 *     <li>{@link #onError(Throwable)}：处理发布者发送的错误。</li>
 *     <li>{@link #onComplete()}：处理发布者完成数据发送。</li>
 * </ul>
 * <p>
 * 示例：
 * <pre>
 * Subscriber<String> subscriber = new Subscriber<>() {
 *     public void onSubscribe(Subscription s) {
 *         s.request(1); // 请求一个数据
 *     }
 *     public void onNext(String t) {
 *         System.out.println("Received: " + t);
 *         s.request(1); // 请求下一个数据
 *     }
 *     public void onError(Throwable t) {
 *         System.err.println("Error: " + t.getMessage());
 *     }
 *     public void onComplete() {
 *         System.out.println("Completed");
 *     }
 * };
 * </pre>
 *
 * @param <T> 接收的数据类型
 */
public interface Subscriber<T> {
    /**
     * 在订阅发布者后执行。
     * <p>
     * 该方法用于接收订阅关系，并通过 {@link Subscription} 对象控制数据流的请求。
     * 订阅者可以通过调用 {@link Subscription#request(long)} 方法请求数据，
     * 或者调用 {@link Subscription#cancel()} 方法取消订阅。
     *
     * @param s 订阅关系
     */
    public void onSubscribe(Subscription s);

    /**
     * 接收发布者发送的下一个数据。
     * <p>
     * 该方法会在订阅者调用 {@link Subscription#request(long)} 方法请求数据后被调用。
     * 订阅者可以通过调用 {@link Subscription#request(long)} 方法继续请求数据，
     * 或者调用 {@link Subscription#cancel()} 方法取消订阅。
     *
     * @param t 数据
     */
    public void onNext(T t);

    /**
     * 处理发布者发送的错误。
     * <p>
     * 该方法会在发布者遇到错误时被调用。订阅者可以通过此方法处理错误。
     *
     * @param t 错误
     */
    public void onError(Throwable t);

    /**
     * 处理发布者完成数据发送。
     * <p>
     * 该方法会在发布者完成所有数据发送后被调用。订阅者可以通过此方法处理完成事件。
     */
    public void onComplete();
}




/**
 * 用于发布者与订阅者之间的通信接口。
 * <p>
 * 该接口用于实现背压机制（Backpressure），允许订阅者告诉发布者需要多少数据。
 * <p>
 * 示例：
 * <pre>
 * Subscription subscription = new Subscription() {
 *     public void request(long n) {
 *         // 请求 n 个数据
 *     }
 *     public void cancel() {
 *         // 取消订阅
 *     }
 * };
 * </pre>
 */
public interface Subscription {
    /**
     * 请求数据。
     * <p>
     * 订阅者通过此方法请求数据。参数 {@code n} 表示请求的数据数量。
     * 发布者会根据请求的数量发送数据。
     *
     * @param n 请求的数据数量
     */
    public void request(long n);

    /**
     * 取消订阅。
     * <p>
     * 订阅者通过此方法取消订阅。取消后，发布者将停止发送数据。
     */
    public void cancel();
}




/**
 * 处理器接口。
 * <p>
 * 用于处理发布者（Publisher）发布的消息，并将处理后的消息传递给订阅者（Subscriber）。
 * 处理器既是一个订阅者（Subscriber），也是一个发布者（Publisher）。
 * <p>
 * 示例：
 * <pre>
 * Processor<String, String> processor = new Processor<>() {
 *     public void onSubscribe(Subscription s) {
 *         s.request(1);
 *     }
 *     public void onNext(String t) {
 *         // 处理消息
 *         String processed = t.toUpperCase();
 *         // 将处理后的消息传递给订阅者
 *         subscriber.onNext(processed);
 *     }
 *     public void onError(Throwable t) {
 *         subscriber.onError(t);
 *     }
 *     public void onComplete() {
 *         subscriber.onComplete();
 *     }
 *     public void subscribe(Subscriber<? super String> s) {
 *         this.subscriber = s;
 *     }
 * };
 * </pre>
 *
 * @param <T> 接收的数据类型
 * @param <R> 发送的数据类型
 */
public interface Processor<T, R> extends Subscriber<T>, Publisher<R> {
}

/**
 * 使用 {@link MonoSink} 来创建一个 {@link Mono}。
 * <p>
 * 通过 {@link MonoSink} 提供的 API（如 {@link MonoSink#success(Object)} 或 {@link MonoSink#error(Throwable)}），
 * 可以手动控制数据的发布或错误的传播。
 * <p>
 * 示例：
 * <pre>
 * Mono<String> mono = Mono.create(sink -> {
 *     if (someCondition) {
 *         sink.success("Hello, World!");
 *     } else {
 *         sink.error(new IllegalArgumentException("Condition not met"));
 *     }
 * });
 * </pre>
 *
 * @param <T> 数据类型
 * @return 一个新的 {@link Mono} 实例
 */
Mono.create();

/**
 * 从一个 {@link Optional} 对象或 {@code null} 对象中创建一个 {@link Mono}。
 * <p>
 * 如果 {@link Optional} 对象中包含值，或者对象不为 {@code null}，则 Mono 序列会产生对应的元素；
 * 否则，返回一个空的 Mono。
 * <p>
 * 示例：
 * <pre>
 * Optional<String> optional = Optional.of("Value");
 * Mono<String> mono = Mono.justOrEmpty(optional);
 * </pre>
 *
 * @param <T> 数据类型
 * @param optional 包含值的 {@link Optional} 对象或 {@code null}
 * @return 一个新的 {@link Mono} 实例
 */
Mono.justOrEmpty();

/**
 * 创建一个只包含错误消息的 {@link Mono}。
 * <p>
 * 该方法用于直接传播一个错误信号，而不是正常的数据。
 * <p>
 * 示例：
 * <pre>
 * Mono<String> mono = Mono.error(new RuntimeException("Something went wrong"));
 * </pre>
 *
 * @param <T> 数据类型（虽然不会产生数据，但需要指定类型）
 * @param error 要传播的错误
 * @return 一个新的 {@link Mono} 实例
 */
Mono.error();

/**
 * 创建一个不包含任何消息通知的 {@link Mono}。
 * <p>
 * 该 Mono 不会发出任何数据或完成信号，主要用于测试或特殊场景。
 * <p>
 * 示例：
 * <pre>
 * Mono<String> mono = Mono.never();
 * </pre>
 *
 * @param <T> 数据类型
 * @return 一个新的 {@link Mono} 实例
 */
Mono.never();

/**
 * 在指定的延迟时间之后，创建一个 {@link Mono}，产生数字 0 作为唯一值。
 * <p>
 * 该方法用于在延迟指定时间后发出一个信号，常用于模拟异步操作的延迟。
 * <p>
 * 示例：
 * <pre>
 * Mono<Long> mono = Mono.delay(Duration.ofSeconds(2));
 * </pre>
 *
 * @param delay 延迟时间
 * @return 一个新的 {@link Mono} 实例
 */
Mono.delay();

/**
 * 创建一个包含单个元素的 {@link Mono}。
 * <p>
 * 该方法用于创建一个包含非 {@code null} 数据的 Mono，声明的参数就是数据流的元素。
 * 创建出来的 Mono 序列在发布这些元素之后会自动结束。
 * <p>
 * 示例：
 * <pre>
 * Mono<String> mono = Mono.just("Hello, World!");
 * </pre>
 *
 * @param <T> 数据类型
 * @param value 数据流中的元素
 * @return 一个新的 {@link Mono} 实例
 */
Mono.just();

/**
 * 从一个回调函数中创建一个 {@link Mono}。
 * <p>
 * 该方法允许通过一个回调函数来生成数据，回调函数的返回值将作为 Mono 的输出。
 * <p>
 * 示例：
 * <pre>
 * Mono<String> mono = Mono.fromCallable(() -> {
 *     // 模拟一些计算或操作
 *     return "Result";
 * });
 * </pre>
 *
 * @param <T> 数据类型
 * @param callable 回调函数
 * @return 一个新的 {@link Mono} 实例
 */
Mono.fromCallable();

/**
 * 从一个 {@link CompletionStage} 对象中创建一个 {@link Mono}。
 * <p>
 * 该方法用于将异步任务的结果包装为一个 Mono，从而可以与其他响应式流进行组合。
 * <p>
 * 示例：
 * <pre>
 * CompletionStage<String> stage = CompletableFuture.supplyAsync(() -> "Async Result");
 * Mono<String> mono = Mono.fromCompletionStage(stage);
 * </pre>
 *
 * @param <T> 数据类型
 * @param stage 异步任务的 {@link CompletionStage} 对象
 * @return 一个新的 {@link Mono} 实例
 */
Mono.fromCompletionStage();

/**
 * 从一个 {@link CompletableFuture} 对象中创建一个 {@link Mono}。
 * <p>
 * 该方法用于将异步任务的结果包装为一个 Mono，从而可以与其他响应式流进行组合。
 * <p>
 * 示例：
 * <pre>
 * CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Async Result");
 * Mono<String> mono = Mono.fromFuture(future);
 * </pre>
 *
 * @param <T> 数据类型
 * @param future 异步任务的 {@link CompletableFuture} 对象
 * @return 一个新的 {@link Mono} 实例
 */
Mono.fromFuture();

/**
 * 从一个 {@link Runnable} 对象中创建一个 {@link Mono}。
 * <p>
 * 该方法用于将一个异步任务包装为一个 Mono，任务执行完成后，Mono 会发出一个完成信号。
 * <p>
 * 示例：
 * <pre>
 * Mono<Void> mono = Mono.fromRunnable(() -> {
 *     // 执行一些异步任务
 *     System.out.println("Task executed");
 * });
 * </pre>
 *
 * @param <T> 数据类型（通常为 {@link Void}，因为 Runnable 不返回值）
 * @param runnable 异步任务的 {@link Runnable} 对象
 * @return 一个新的 {@link Mono} 实例
 */
Mono.fromRunnable();

/**
 * 从一个 {@link Supplier} 对象中创建一个 {@link Mono}。
 * <p>
 * 该方法用于将一个提供者包装为一个 Mono，提供者的结果将作为 Mono 的输出。
 * <p>
 * 示例：
 * <pre>
 * Mono<String> mono = Mono.fromSupplier(() -> {
 *     // 提供一个值
 *     return "Provided Value";
 * });
 * </pre>
 *
 * @param <T> 数据类型
 * @param supplier 提供者
 * @return 一个新的 {@link Mono} 实例
 */
Mono.fromSupplier();

/**
 * 从数组创建一个 {@link Flux}，该 Flux 会依次发出数组中的元素。
 * <p>
 * 如果数组为空，则返回一个空的 Flux；如果数组只有一个元素，则返回一个包含单个元素的 Flux；
 * 如果数组包含多个元素，则通过 {@link FluxArray} 来实现元素的发布。
 *
 * @param array 数组，用于提供数据源
 * @param <T> 数组中元素的类型，也是返回的 Flux 中元素的类型
 * @return 一个新的 {@link Flux} 实例
 */
public static <T> Flux<T> fromArray(T[] array) {
    // 如果数组为空，返回一个空的 Flux
    if (array.length == 0) {
        return empty();
    }
    // 如果数组只有一个元素，使用 just 方法直接返回一个包含单个元素的 Flux
    if (array.length == 1) {
        return just(array[0]);
    }
    // 对于包含多个元素的数组，返回一个新的 FluxArray 实例
    return onAssembly(new FluxArray<>(array));
}

/**
 * 用于从数组中发布数据的内部类。
 * <p>
 * 该类实现了 {@link Flux} 接口，并通过 {@link SourceProducer} 提供数据源。
 * 它支持普通订阅者和条件订阅者两种模式。
 */
final class FluxArray<T> extends Flux<T> implements Fuseable, SourceProducer<T> {
    /**
     * 存储数组数据。
     */
    final T[] array;

    /**
     * 构造函数，使用可变参数接收数组。
     * <p>
     * 数组不能为空，否则会抛出 {@link NullPointerException}。
     *
     * @param array 数组，用于提供数据源
     */
    @SafeVarargs
    public FluxArray(T... array) {
        this.array = Objects.requireNonNull(array, "数组不能为空");
    }

    /**
     * 订阅方法，根据订阅者类型选择不同的订阅逻辑。
     * <p>
     * 如果数组为空，直接完成订阅；如果者订阅是 {@link ConditionalSubscriber} 类型，
     * 则使用 {@link ArrayConditionalSubscription}；否则使用普通的 {@link ArraySubscription}。
     *
     * @param s 订阅者
     * @param array 数组数据
     */
    @SuppressWarnings("unchecked")
    public static <T> void subscribe(CoreSubscriber<? super T> s, T[] array) {
        // 如果数组为空，直接完成订阅
        if (array.length == 0) {
            Operators.complete(s);
            return;
        }
        // 如果订阅者是 ConditionalSubscriber 类型，使用 ArrayConditionalSubscription
        if (s instanceof ConditionalSubscriber) {
            s.onSubscribe(new ArrayConditionalSubscription<>((ConditionalSubscriber<? super T>) s, array));
        }
        // 否则使用普通的 ArraySubscription
        else {
            s.onSubscribe(new ArraySubscription<>(s, array));
        }
    }

    /**
     * 普通订阅者的实现。
     * <p>
     * 该类实现了 {@link InnerProducer} 和 {@link SynchronousSubscription} 接口，
     * 用于处理普通订阅者的请求和取消操作。
     */
    static final class ArraySubscription<T> implements InnerProducer<T>, SynchronousSubscription<T> {
        /**
         * 实际的订阅者。
         */
        final CoreSubscriber<? super T> actual;
        /**
         * 存储数组数据。
         */
        final T[] array;
        /**
         * 当前索引。
         */
        int index;
        /**
         * 记录是否取消订阅。
         */
        volatile boolean cancelled;
        /**
         * 记录请求的元素数量。
         */
        volatile long requested;

        /**
         * 用于线程安全地更新 requested 字段的原子更新器。
         */
        @SuppressWarnings("rawtypes")
        static final AtomicLongFieldUpdater<ArraySubscription> REQUESTED =
                AtomicLongFieldUpdater.newUpdater(ArraySubscription.class, "requested");

        /**
         * 构造函数，初始化订阅者和数组。
         *
         * @param actual 实际的订阅者
         * @param array 数组数据
         */
        ArraySubscription(CoreSubscriber<? super T> actual, T[] array) {
            this.actual = actual;
            this.array = array;
        }

        /**
         * 请求方法，用于订阅者请求元素。
         * <p>
         * 根据请求数量，选择快速路径或慢速路径来发送元素。
         *
         * @param n 请求的元素数量
         */
        @Override
        public void request(long n) {
            // 验证请求的合法性，请求数量必须为正数
            if (Operators.validate(n)) {
                // 增加请求的元素数量
                if (Operators.addCap(REQUESTED, this, n) == 0) {
                    // 如果请求数量为 Long.MAX_VALUE，表示请求所有元素，走快速路径
                    if (n == Long.MAX_VALUE) {
                        fastPath();
                    }
                    // 否则走慢速路径
                    else {
                        slowPath(n);
                    }
                }
            }
        }

        /**
         * 快速路径：请求所有元素。
         * <p>
         * 遍历数组，逐个发送元素，直到取消订阅或所有元素发送完毕。
         */
        private void fastPath() {
            // 遍历数组，逐个发送元素
            for (T t : array) {
                if (cancelled) {
                    return; // 如果取消订阅，停止发送
                }
                actual.onNext(t); // 发送元素
            }
            if (cancelled) {
                return; // 再次检查是否取消
            }
            Operators.complete(actual); // 发送完成信号
        }

        /**
         * 慢速路径：根据请求数量逐个发送元素。
         * <p>
         * 遍历数组，发送请求数量的元素，直到取消订阅或所有元素发送完毕。
         *
         * @param n 请求的元素数量
         */
        private void slowPath(long n) {
            // 遍历数组，发送请求数量的元素
            for (int i = 0; i < n && index < array.length; i++) {
                if (cancelled) {
                    return; // 如果取消订阅，停止发送
                }
                actual.onNext(array[index++]); // 发送当前索引的元素并递增索引
            }
            if (index == array.length) {
                if (cancelled) {
                    return; // 再次检查是否取消
                }
                Operators.complete(actual); // 如果所有元素发送完毕，发送完成信号
            }
        }

        /**
         * 取消订阅方法。
         * <p>
         * 标记为已取消，停止后续操作。
         */
        @Override
        public void cancel() {
            cancelled = true; // 标记为已取消
        }

        /**
         * 获取当前索引。
         *
         * @return 当前索引值
         */
        @Override
        public int get() {
            return index;
        }

        /**
         * 设置当前索引。
         *
         * @param index 新的索引值
         */
        @Override
        public void set(int index) {
            this.index = index;
        }

        /**
         * 增加当前索引。
         */
        @Override
        public void increment() {
            index++;
        }
    }

    /**
     * 条件订阅者的实现（未展开，可根据需要补充）。
     * <p>
     * 该类用于处理条件订阅者的请求和取消操作。
     */
    static final class ArrayConditionalSubscription<T> implements InnerProducer<T>, SynchronousSubscription<T> {
        /**
         * 实际的条件订阅者。
         */
        final ConditionalSubscriber<? super T> actual;
        /**
         * 存储数组数据。
         */
        final T[] array;

        /**
         * 构造函数，初始化条件订阅者和数组。
         *
         * @param actual 实际的条件订阅者
         * @param array 数组数据
         */
        ArrayConditionalSubscription(ConditionalSubscriber<? super T> actual, T[] array) {
            this.actual = actual;
            this.array = array;
        }

        // 条件订阅者的具体实现逻辑（可根据需要补充）
    }
}

   Flux.error(new RuntimeException());

    Flux.error(new RuntimeException());        
    Flux.just();        
    Flux.empty();

  Flux.just(1, 2, 3).subscribe(System.out::println);

Flux.error(new Exception("error"))
		.subscribe(System.out::println,System.err::println
		,() -> System.out.println("Completed!"));

import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;

/**
 * 示例代码：使用 Reactor 的 {@link Flux} 和 {@link BaseSubscriber} 来演示响应式编程的基本流程。
 * <p>
 * 该代码展示了如何创建一个简单的响应式流，如何通过自定义订阅者（BaseSubscriber）来控制数据流的请求和消费，
 * 并通过背压机制（Backpressure）来管理数据的发送。
 * <p>
 * 主要步骤：
 * <ol>
 *     <li>使用 {@link Flux#just(Object...)} 创建一个包含两个元素的响应式流。</li>
 *     <li>通过 {@link Flux#doOnRequest(Consumer)} 在请求数据时打印日志。</li>
 *     <li>自定义订阅者（BaseSubscriber），重写 {@link BaseSubscriber#hookOnSubscribe(Subscription)} 和
 *     {@link BaseSubscriber#hookOnNext(Object)} 方法来控制请求和消费逻辑。</li>
 *     <li>调用 {@link Flux#subscribe(Subscriber)} 方法启动订阅。</li>
 * </ol>
 */
public class ReactiveExample {
    public static void main(String[] args) {
        // 创建一个包含两个元素的响应式流
        Flux.just(1, 2)
                // 在请求数据时打印日志
                .doOnRequest(s -> System.out.println("请求了 " + s + " 个元素"))
                // 自定义订阅者
                .subscribe(new BaseSubscriber<Integer>() {
                    /**
                     * 在订阅开始时调用。
                     * <p>
                     * 该方法用于初始化订阅关系，并请求第一个元素。
                     */
                    @Override
                    protected void hookOnSubscribe(Subscription subscription) {
                        System.out.println("订阅开始了，我要请求几个元素");
                        // 请求第一个元素
                        request(1);
                    }

                    /**
                     * 在接收到一个元素时调用。
                     * <p>
                     * 该方法用于处理接收到的元素，并请求下一个元素。
                     *
                     * @param value 接收到的元素
                     */
                    @Override
                    protected void hookOnNext(Integer value) {
                        System.out.println("收到一个元素：" + value + "，下一次请求几个元素");
                        // 请求下一个元素
                        request(1);
                    }
                });
    }
}

import org.junit.Test;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * 测试类：将同步方法转换为异步执行的示例。
 * <p>
 * 该类展示了如何使用 Reactor 的 {@link Mono} 和 {@link Schedulers} 将同步方法转换为异步执行，
 * 并通过 {@link CountDownLatch} 等待异步任务完成。
 * <p>
 * 主要步骤：
 * <ol>
 *     <li>定义一个同步方法 {@link #syncMethod()}，该方法模拟耗时操作。</li>
 *     <li>在测试方法中，使用 {@link Mono#fromCallable(Callable)} 将同步方法包装为响应式流。</li>
 *     <li>通过 {@link Mono#subscribeOn(Scheduler)} 指定在弹性线程池中执行同步方法。</li>
 *     <li>使用 {@link CountDownLatch} 等待异步任务完成。</li>
 * </ol>
 */
public class AsyncTest {

    /**
     * 同步方法，模拟耗时操作。
     * <p>
     * 该方法会阻塞当前线程 2 秒，然后返回一个整数值。
     * <p>
     * 注意：在实际应用中，应尽量避免在主线程中调用此类同步方法，以免阻塞主线程。
     *
     * @return 返回一个整数值
     */
    private Integer syncMethod() {
        try {
            // 模拟耗时操作
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return 123456;
    }

    /**
     * 测试方法：将同步方法转换为异步执行。
     * <p>
     * 该方法通过 Reactor 的 {@link Mono} 和 {@link Schedulers} 将同步方法转换为异步执行，
     * 并通过 {@link CountDownLatch} 等待异步任务完成。
     */
    @Test
    public void switchSyncToAsyncTest() {
        // 创建一个计数器，用于等待异步任务完成
        CountDownLatch countDownLatch = new CountDownLatch(1);

        // 将同步方法包装为响应式流
        Mono.fromCallable(this::syncMethod)
                // 指定在弹性线程池中执行同步方法
                .subscribeOn(Schedulers.elastic())
                // 订阅响应式流，处理结果
                .subscribe(
                        System.out::println, // 处理正常结果
                        null,               // 处理错误（此处未实现）
                        countDownLatch::countDown // 在任务完成后减少计数器
                );

        try {
            // 等待异步任务完成
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

 Flux.just(1,2,0).map(v -> 2 / v)
 .onErrorReturn(0).map(v - > v * 2)
 .subscribe(System.out::println,System.err::println);

        Flux.just(1,2,0)                 
        //调用redis服务获取数据                
        .flatMap(id -> redisService.getUserByid(id))                
        //当发生异常的时候，从DB用户服务获取数据                
        .onErrorResume(v -> userService.getUserByCache(id));

      Flux.just(1,2,0).flatMap(id -> getUserByid(id))
      .onErrorMap(v -> new CustomizeExcetion("服务器开小差了", v));

        Flux.just(1,2,0)                
        .flatMap(id -> getUserByid(id))                
        .doOnError(e -> Log.error("this occur something error"))
        .onErrorMap(v -> new CustomizeExcetion("服务器开小差了", v));

        Flux.just(1,2,0)
        .flatMap(id -> getUserByid(id))                
        .doOnError(e -> Log.error("this occur something error"))                
        .onErrorMap(v->new CustomizeExcetion("服务器开小差了",v))                
        .doFinally(System.out.println("我会确保做一些事情"));

        Flux.using(() -> createHttpClient(), 
			        client -> Flux.just(client.sendRequest()), createHttpClient::close         );

      Flux.just(1,2,0)              
      .map(v->2/v)              
      .retry(1)              
      .subscribe(System.out::println,System.err::println);

    @Test    
    public void reactorTest() {        
	     StepVerifier.create(Flux.just(1,2)) //1              
	     .expectNext(1,2) //2               
	     .expectComplete() //3               
	     .verify(); //4     
    }

StepVerifier.withVirtualTime(() -> Mono.delay(Duration.ofDays(1)))    
.expectSubscription()   // 1    
.expectNoEvent(Duration.ofDays(1))  // 2    
.expectNext(0L)    
.verifyComplete();  // 3

    @Test    
    public void reactorMapTest() {        
	    StepVerifier.create(Flux.just(1,2)                
	    .map(v -> v + 1))                
	    .expectNext(2,3)                
	    .expectComplete()                
	    .verify();    
    }

    @Test    
    public void reactorFlatMapTest() {        
	    StepVerifier.create(Flux.just("crabman","is","hero")                
	    .flatMap(v -> Flux.fromArray(v.split("")))                
	    .doOnNext(System.out::println))                
	    .expectNextCount(13)                
	    .verifyComplete();    
    }

    @Test    
    public void reactorFilterTest() {        
	     StepVerifier.create(Flux.just(1,2)                
	     .map(v -> v + 1)
	     .filter(s -> s != 2)                
	     .doOnNext(System.out::println))                
	     .expectNext(3)                
	     .expectComplete()                
	     .verify();    
    }

    @Test    
    public void reactorZipTest() {        
		    //定义一个Flux流        
		    Flux<String> stringFlux = Flux.just("a", "b", "c", "d");        
		    //这里使用计时器，因为在单元测试里面，可能元素没执行完，他就会直接返回        
		    CountDownLatch countDownLatch = new CountDownLatch(1);  
		    // 2        
		    Flux.zip(stringFlux,Flux.interval(Duration.ofSeconds(1)))                
		    .subscribe(t -> System.out.println(t.getT1()),System.err::println ,countDownLatch::countDown);  
		    
		        try {            
		          countDownLatch.await();        
	          } catch (InterruptedException e) {            
	          e.printStackTrace();        
	          }    
      }

public class FlowExample {
    public static void main(String[] args) {
        Publisher<Integer> publisher = subscriber -> {
            subscriber.onNext(1);
            subscriber.onNext(2);
            subscriber.onNext(3);
            subscriber.onComplete();
        };

        Subscriber<Integer> subscriber = new Subscriber<>() {
            private Subscription subscription;

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

            @Override
            public void onError(Throwable throwable) {
                System.err.println("Error: " + throwable.getMessage());
            }

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

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

        publisher.subscribe(subscriber);
    }
}

import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;

@RestController
public class ReactiveController {

    private static final Logger log = Logger.getLogger(ReactiveController.class.getName());

    /**
     * 模拟耗时操作，阻塞 5 秒钟。
     * <p>
     * 该方法用于模拟一个同步的、耗时的操作，返回一个字符串。
     *
     * @return 返回一个字符串
     */
    private String createStr() {
        try {
            // 阻塞 5 秒
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
            // 忽略中断异常
        }
        return "……^ - ^";
    }

    /**
     * 原生 MVC 请求处理方法。
     * <p>
     * 该方法使用传统的同步方式处理请求，调用耗时方法 {@link #createStr()}，
     * 并记录请求的开始时间和结束时间。
     *
     * @return 返回处理结果
     */
    @GetMapping("/mvc")
    public String mvc() {
        long millis = System.currentTimeMillis();
        log.info("请求1: " + millis);
        // 调用耗时方法
        String result = createStr();
        log.info("结束1: " + (System.currentTimeMillis() - millis) + " ms");
        return result;
    }

    /**
     * WebFlux 请求处理方法。
     * <p>
     * 该方法使用响应式编程的方式处理请求，将耗时方法包装为响应式流，
     * 并记录请求的开始时间和结束时间。
     * <p>
     * 与原生 MVC 方法相比，该方法不会阻塞主线程，从而提高系统的并发处理能力。
     *
     * @return 返回一个响应式流 {@link Mono<String>}
     */
    @GetMapping("/flux")
    public Mono<String> flux() {
        long millis = System.currentTimeMillis();
        log.info("请求2: " + millis);
        // 将耗时方法包装为响应式流，并在弹性线程池中执行
        Mono<String> result = Mono.fromSupplier(this::createStr)
                                   .subscribeOn(Schedulers.elastic());
        log.info("结束2: " + (System.currentTimeMillis() - millis) + " ms");
        return result;
    }
}

import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;

import java.time.Duration;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;

@RestController
public class TimingController {

    /**
     * 定时返回 0 到 n 个元素。
     * <p>
     * 该方法通过 Reactor 的 {@link Flux} 创建一个定时生成数据的流。
     * 每隔 1 秒生成一个字符串元素，总共生成 4 个元素。
     * <p>
     * 注意：该方法需要指定响应的媒体类型为 {@link MediaType#TEXT_EVENT_STREAM_VALUE}，
     * 以便支持 Server-Sent Events (SSE) 的流式响应。
     *
     * @return 一个包含定时生成的字符串的响应式流 {@link Flux<String>}
     */
    @GetMapping(value = "/timing", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<String> timing() {
        // 创建一个 Flux，每隔 1 秒生成一个字符串元素
        Flux<String> result = Flux.fromStream(IntStream.range(1, 5).mapToObj(i -> {
            try {
                // 模拟耗时操作
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                // 忽略中断异常
            }
            return "大内密探00" + i;
        }));

        return result;
    }
}

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.http.MediaType;
import org.springframework.web.reactive.function.server.RouterFunction;
import org.springframework.web.reactive.function.server.RouterFunctions;
import org.springframework.web.reactive.function.server.ServerResponse;

import static org.springframework.web.reactive.function.server.RequestPredicates.GET;
import static org.springframework.web.reactive.function.server.RequestPredicates.accept;

@Configuration
public class RouterConfig {

    @Bean
    public RouterFunction<ServerResponse> route(Handler handler) {
        return RouterFunctions
               .route(GET("/hello").and(accept(MediaType.TEXT_PLAIN)), handler::hello);
    }
}

import org.springframework.http.MediaType;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.server.ServerRequest;
import org.springframework.web.reactive.function.server.ServerResponse;
import reactor.core.publisher.Mono;

@Component
public class Handler {

    public Mono<ServerResponse> hello(ServerRequest request) {
        return ServerResponse.ok().contentType(MediaType.TEXT_PLAIN)
               .body(Mono.just("Hello, WebFlux!"), String.class);
    }
}

import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;

public class WebClientExample {

    public static void main(String[] args) {
        WebClient webClient = WebClient.create("https://example.com");
        Mono<String> response = webClient.get()
               .uri("/api/data")
               .retrieve()
               .bodyToMono(String.class);

        response.subscribe(System.out::println);
    }
}

import org.springframework.boot.web.error.ErrorAttributeOptions;
import org.springframework.boot.web.reactive.error.DefaultErrorAttributes;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.server.ServerRequest;

import java.util.Map;

@Component
public class CustomErrorAttributes extends DefaultErrorAttributes {

    @Override
    public Map<String, Object> getErrorAttributes(ServerRequest request, ErrorAttributeOptions options) {
        Map<String, Object> errorAttributes = super.getErrorAttributes(request, options);
        // 自定义错误属性
        errorAttributes.put("customAttribute", "Custom Value");
        return errorAttributes;
    }
}

/**
 * 查询处理器，用于处理请求并返回对应的处理器。
 * <p>
 * 该方法是 Spring WebFlux 中的一个核心方法，用于根据请求匹配路由函数并返回对应的处理器。
 * 如果没有匹配的路由函数，则返回一个空的 {@link Mono}。
 *
 * @param exchange 当前的服务器 Web 交换对象，包含了请求和响应的信息
 * @return 返回一个包含处理器的 {@link Mono}，或者一个空的 {@link Mono}
 */
@Override
protected Mono<?> getHandlerInternal(ServerWebExchange exchange) {
    // 检查路由函数是否存在
    if (this.routerFunction != null) {
        // 创建一个 ServerRequest 对象，并绑定当前的交换对象和消息读取器
        ServerRequest request = ServerRequest.create(exchange, this.messageReaders);

        // 使用路由函数匹配请求，并返回对应的处理器
        return this.routerFunction.route(request)
                // 在匹配到处理器后，设置一些属性到交换对象的属性中
                .doOnNext(handler -> setAttributes(exchange.getAttributes(), request, handler));
    } else {
        // 如果没有路由函数，则返回一个空的 Mono
        return Mono.empty();
    }
}

import org.springframework.web.reactive.function.server.HandlerFunction;
import org.springframework.web.reactive.function.server.ServerRequest;
import reactor.core.publisher.Mono;

/**
 * 路由方法，根据请求匹配路由谓词并返回对应的处理器函数。
 * <p>
 * 该方法是路由功能的核心实现，用于检查请求是否匹配预定义的路由谓词。
 * 如果匹配成功，则返回一个包含处理器函数的 {@link Mono}；如果不匹配，则返回一个空的 {@link Mono}。
 *
 * @param <T> 处理器函数的返回类型
 */
public class RouterFunction<T> {

    private final HandlerFunction<T> handlerFunction;
    private final java.util.function.Predicate<ServerRequest> predicate;
    private final org.slf4j.Logger logger;

    /**
     * 构造函数，初始化路由谓词和处理器函数。
     *
     * @param predicate 路由谓词，用于匹配请求
     * @param handlerFunction 匹配成功时的处理器函数
     * @param logger 日志记录器
     */
    public RouterFunction(java.util.function.Predicate<ServerRequest> predicate,
                          HandlerFunction<T> handlerFunction,
                          org.slf4j.Logger logger) {
        this.predicate = predicate;
        this.handlerFunction = handlerFunction;
        this.logger = logger;
    }

    /**
     * 根据请求匹配路由谓词并返回对应的处理器函数。
     * <p>
     * 该方法检查请求是否匹配预定义的路由谓词。如果匹配成功，返回一个包含处理器函数的 {@link Mono}；
     * 如果不匹配，返回一个空的 {@link Mono}。
     *
     * @param request 当前的请求对象
     * @return 一个包含处理器函数的 {@link Mono}，或者一个空的 {@link Mono}
     */
    public Mono<HandlerFunction<T>> route(ServerRequest request) {
        // 检查路由谓词是否匹配请求
        if (this.predicate.test(request)) {
            // 如果匹配成功，记录日志（如果日志级别为 TRACE）
            if (logger.isTraceEnabled()) {
                String logPrefix = request.exchange().getLogPrefix();
                logger.trace(logPrefix + String.format("Matched %s", this.predicate));
            }
            // 返回包含处理器函数的 Mono
            return Mono.just(this.handlerFunction);
        } else {
            // 如果不匹配，返回一个空的 Mono
            return Mono.empty();
        }
    }
}