Flux.concat 使用说明书

public static <T> Flux<T> concat(Iterable<? extends Publisher<? extends T>> sources)

Concatenate all sources provided in an Iterable, forwarding elements emitted by the sources downstream.

连接可迭代集合中提供的所有源，将源发出的元素转发到下游。

Concatenation is achieved by sequentially subscribing to the first source then waiting for it to complete before subscribing to the next, and so on until the last source completes. Any error interrupts the sequence immediately and is forwarded downstream.

连接是通过顺序订阅第一个源来实现的，然后等待其完成后再订阅下一个源，以此类推，直到最后一个源完成。任何错误都会立即中断序列并转发到下游。

Type Parameters:

T - The type of values in both source and output sequences

T - 源和输出序列中值的类型

Parameters:

sources - The Iterable of Publisher to concatenate

sources - 要连接的发布者的可迭代集合

Returns:

a new Flux concatenating all source sequences

一个新的 Flux，连接所有源序列。

concat(Iterable<? extends Publisher<? extends T>> sources) 是一个静态方法，它可以将多个 Publisher 逐个按顺序连接为一个 Flux。

这个操作将依次订阅每个 Publisher，等前一个 Publisher 完成后，再订阅下一个。所有的 Publisher 中发出的元素将串联为一个 Flux 流。

参数解释：

sources：一个包含多个 Publisher 的 Iterable。这些 Publisher 发出的元素类型为 T，concat 会将它们按顺序连接。

特点：

按顺序连接：与 merge 不同，concat 会等到前一个 Publisher 完成后，才开始订阅下一个。即每个 Publisher 都会按顺序一个一个执行，不会交错。
顺序保证：由于是按顺序连接，这个方法可以保证每个 Publisher 的发出顺序不会被打乱。

示例代码：

java

public static void main(String[] args) {
    // 定义多个 Publisher（Flux）Flux<String> flux1 = Flux.just("A", "B", "C");Flux<String> flux2 = Flux.just("D", "E");Flux<String> flux3 = Flux.just("F", "G", "H");

    // 使用 concat 将多个 Flux 按顺序连接Flux<String> result = concat(Arrays.asList(flux1, flux2, flux3));

    // 订阅并打印结果
    result.subscribe(System.out::println);
}

// 定义 concat 方法
public static <T> Flux<T> concat(Iterable<? extends Publisher<? extends T>> sources) {
    return Flux.concat(sources);
}

输出结果：

A
B
C
D
E
F
G
H

代码详解：

flux1、flux2、flux3：三个不同的 Flux，分别发出一组字符串。
concat 方法：将 flux1、flux2 和 flux3 组合为一个 Flux，每个 Flux 的值依次发出。concat 确保第一个 Flux 完成后，才会订阅并发出第二个 Flux，依此类推。
订阅和输出：合并后的 Flux 将所有的元素按顺序打印出来。

场景应用：

顺序执行多个异步任务：适用于需要按顺序执行多个任务的场景，特别是当前一个任务必须完成后才能执行下一个任务时。
日志处理或事件流：例如，在多个日志流、事件流的处理过程中，要求严格按时间顺序来执行和处理。

其他用法：

如果 sources 中有 Mono，它们也可以按顺序连接，并且由于 Mono 只发出一个元素，能确保顺序处理。

总结：

concat 是一个用于按顺序连接多个 Publisher 的工具，保证它们一个接一个发出数据，不会交错。
使用这个方法可以确保多个异步操作的顺序，适用于那些需要顺序严格处理的数据流合并场景。

concatWithValues

@SafeVarargs
public final Flux<T> concatWithValues(T... values)

Concatenates the values to the end of the Flux

将值连接到 Flux 的末尾。

Parameters:

values - The values to concatenate

values - 要连接的值

Returns:

a new Flux concatenating all source sequences

一个新的 Flux，连接所有源序列。

concatWithValues(T... values) 是一个 Flux 的实例方法，用于将当前 Flux 实例与多个单独的值按顺序连接起来。

这个方法接收可变参数 values，表示要附加到当前 Flux 后的值，形成一个新的 Flux，它首先发出原始 Flux 的元素，之后依次发出 values 中的值。

特点：

顺序连接：先发出当前 Flux 中的元素，等当前流完成后，再发出 values 中的值。
终止信号后发出值：concatWithValues 是在原来的 Flux 正常完成（即发出 onComplete 信号）后，再发出附加的值。

参数解释：

values：需要在当前 Flux 完成后发出的附加值，类型为 T，可以传入一个或多个值。

示例代码：

java

public static void main(String[] args) {
    // 创建一个 Flux，包含多个元素
    Flux<String> flux = Flux.just("A", "B", "C");

    // 使用 concatWithValues 方法将 "D", "E", "F" 连接到 Flux 之后
    Flux<String> result = flux.concatWithValues("D", "E", "F");

    // 订阅并打印输出
    result.subscribe(System.out::println);
}

输出结果：

A
B
C
D
E
F

代码详解：

flux：一个发出 "A", "B", "C" 的 Flux。
concatWithValues("D", "E", "F")：将 "D", "E", "F" 这些值附加到 flux 之后，形成新的 Flux。
订阅并输出：首先会打印 flux 中的元素，然后再依次打印 concatWithValues 中的元素。

场景应用：

扩展数据流：可以用于在已有的数据流后附加一些固定的值，比如说在流结束时附加额外的结束标记或者状态信息。
日志或通知系统：在一个数据流处理完后，追加一些结束消息、统计信息等。

其他用法：

假如你想要将多个不同类型的 Flux 或 Mono 合并在一起（类似于 concatWith），也可以考虑使用 concatWithValues 来快速附加固定值。

java

Flux<Integer> numbers = Flux.just(1, 2, 3);
numbers.concatWithValues(4, 5).subscribe(System.out::println);

总结：

concatWithValues 是一种便捷方式，允许你将若干值添加到当前 Flux 结束之后的输出中。
它确保原始 Flux 中的所有数据都已发出，且正常完成后，再发出附加的值，适合需要按顺序扩展数据流的场景。

concat

public static <T> Flux<T> concat(Publisher<? extends Publisher<? extends T>> sources)

Concatenate all sources emitted as an onNext signal from a parent Publisher, forwarding elements emitted by the sources downstream.

将所有作为父发布者的 onNext 信号发出的源连接起来，将源发出的元素转发到下游。

Concatenation is achieved by sequentially subscribing to the first source then waiting for it to complete before subscribing to the next, and so on until the last source completes. Any error interrupts the sequence immediately and is forwarded downstream.

Discard Support:

This operator discards elements it internally queued for backpressure upon cancellation.

Type Parameters:

T - The type of values in both source and output sequences

Parameters:

sources - The Publisher of Publisher to concatenate

Returns:

a new Flux concatenating all inner sources sequences

丢弃支持：

该操作符在取消时会丢弃它内部排队的用于背压的元素。

类型参数：

T - 源和输出序列中值的类型

参数：

sources - 要连接的发布者的发布者

返回：

一个新的 Flux，连接所有内部源序列。

concat(Publisher<? extends Publisher<? extends T>> sources) 是一个静态方法，它可以将一个发出 Publisher 的 Publisher 中的所有元素串联成一个 Flux，并按顺序发出。

与其他 concat 方法不同的是，这个方法的 sources 参数本身是一个发出多个 Publisher 的 Publisher，因此它的功能是先订阅 sources，再依次订阅每个发出的 Publisher，将它们的元素按顺序合并为一个 Flux。

参数解释：

sources：这是一个发出 Publisher 实例的 Publisher。每个 Publisher 都发出元素类型为 T 的数据。concat 会将这些 Publisher 中的数据按顺序串联。

特点：

按顺序连接：当 sources 发出一个 Publisher，concat 会等待这个 Publisher 完成后，才会订阅并处理下一个 Publisher。
顺序处理：不会并发处理这些 Publisher，它保证了严格的顺序，即前一个 Publisher 发出的所有元素必须先处理完，后一个 Publisher 才会被订阅。

使用场景：

动态数据流的顺序处理：适合那些数据流是嵌套的，即 Publisher 发出 Publisher 的情况。可以逐个按顺序处理这些 Publisher，并按顺序输出它们的元素。
异步任务的顺序调度：当多个异步任务之间有明确的顺序要求时，可以使用这种方法，确保在一个任务完成后再开始下一个任务。

示例代码：

java

public static void main(String[] args) {
    // 创建三个 Flux，分别发出不同的值Flux<String> flux1 = Flux.just("A", "B", "C");Flux<String> flux2 = Flux.just("D", "E");
    Flux<String> flux3 = Flux.just("F", "G", "H");

    // 创建一个 Flux 发出这三个 FluxFlux<Flux<String>> sourceFlux = Flux.just(flux1, flux2, flux3);

    // 使用 concat 将这些 Flux 连接为一个按顺序发出元素的 FluxFlux<String> result = concat(sourceFlux);

    // 订阅并打印输出
    result.subscribe(System.out::println);
}

// 定义 concat 方法
public static <T> Flux<T> concat(Publisher<? extends Publisher<? extends T>> sources) {
    return Flux.concat(sources);
}

输出结果：

A
B
C
D
E
F
G
H

代码详解：

flux1、flux2、flux3：这三个 Flux 分别发出不同的字符串。
sourceFlux：这是一个 Flux<Flux<String>>，它发出上面定义的三个 Flux。
concat(sourceFlux)：调用 concat 方法，将发出的多个 Flux 串联为一个 Flux，并按顺序发出元素。
订阅和输出：合并后的 Flux 将所有的元素按顺序输出。

注意事项：

顺序保证：concat 保证了顺序，因此它只会在一个 Publisher 完成后再处理下一个。这意味着如果某个 Publisher 需要较长时间才能完成，后续的 Publisher 将会等待。
适用于顺序性很重要的场景：如果数据之间没有顺序要求，可以考虑使用 merge 来并发处理多个 Publisher，这样可以提高性能，但会失去顺序。

扩展用法：

如果你有多个动态生成的 Publisher，例如每个 Publisher 都是一个远程请求的响应流，这种方法可以确保远程请求按顺序处理，而不会并发发送请求。

总结：

concat(Publisher<? extends Publisher<? extends T>> sources) 可以将一个发出多个 Publisher 的 Publisher 中的所有元素按顺序连接成一个 Flux。
它适用于需要严格按顺序处理多个异步数据源的场景，保证数据的顺序性和依次执行的逻辑。

concat

public static <T> Flux<T> concat(Publisher<? extends Publisher<? extends T>> sources,
                                 int prefetch)

Concatenate all sources emitted as an onNext signal from a parent Publisher, forwarding elements emitted by the sources downstream.4

将所有作为父发布者的 onNext 信号发出的源连接起来，将源发出的元素转发到下游。

Concatenation is achieved by sequentially subscribing to the first source then waiting for it to complete before subscribing to the next, and so on until the last source completes. Any error interrupts the sequence immediately and is forwarded downstream.

订阅第一个源实现，然后等待其完成后再订阅下一个源，以此类推，直到最后一个源完成。任何错误都会立即中断序列并转发到下游。

Discard Support:

This operator discards elements it internally queued for backpressure upon cancellation.

Type Parameters:

T - The type of values in both source and output sequences

Parameters:

sources - The Publisher of Publisher to concatenate

prefetch - the number of Publishers to prefetch from the outer Publisher

Returns:

a new Flux concatenating all inner sources sequences

丢弃支持：

该操作符在取消时会丢弃它内部排队的用于背压的元素。

类型参数：

T - 源和输出序列中值的类型

参数：

sources - 要连接的发布者的发布者

prefetch - 从外部发布者预取的发布者数量

返回：

一个新的 Flux，连接所有内部源序列。

concat(Publisher<? extends Publisher<? extends T>> sources, int prefetch) 是一个静态方法，它将一个发出多个 Publisher 的 Publisher 中的所有元素按顺序连接，并允许通过 prefetch 参数控制预取策略，即在处理过程中能够预取的元素数量。

这个方法与无 prefetch 参数的 concat 方法类似，但增加了对处理流量的控制，可以提高流处理的性能，特别是在流量较大时。

参数解释：

sources：一个 Publisher，它会发出多个 Publisher 实例，每个 Publisher 都会发出类型为 T 的元素。
prefetch：预取数量，表示可以提前请求的元素数量，影响处理速度和资源占用的权衡。prefetch 值越大，预取的数据越多，处理效率也可能越高，但会占用更多内存。

特点：

顺序连接：与其他 concat 方法一样，它会按顺序处理发出的 Publisher，等待一个 Publisher 完成后再处理下一个。
流量控制：prefetch 用于控制流量背压，即在处理过程中可以提前获取的元素数量。默认情况下，concat 会一个一个地请求元素，prefetch 则可以控制一次请求多个元素。
适用于大数据量：当有大量数据需要按顺序处理时，prefetch 可以提高性能，避免频繁的请求和响应开销。

示例代码：

java

public static void main(String[] args) {
    // 创建三个 Flux，分别发出不同的值Flux<String> flux1 = Flux.just("A", "B", "C");Flux<String> flux2 = Flux.just("D", "E");
    Flux<String> flux3 = Flux.just("F", "G", "H");

    // 创建一个 Flux 发出这三个 FluxFlux<Flux<String>> sourceFlux = Flux.just(flux1, flux2, flux3);

    // 使用 concat，将这些 Flux 连接为一个 Flux，并设置 prefetch 为 2Flux<String> result = concat(sourceFlux, 2);

    // 订阅并打印输出
    result.subscribe(System.out::println);
}

// 定义 concat 方法，带有 prefetch 参数
public static <T> Flux<T> concat(Publisher<? extends Publisher<? extends T>> sources, int prefetch) {
    return Flux.concat(sources, prefetch);
}

输出结果：

A
B
C
D
E
F
G
H

代码详解：

flux1、flux2、flux3：三个 Flux，分别发出不同的字符串。
sourceFlux：一个 Flux<Flux<String>>，它发出三个 Flux，即 flux1、flux2 和 flux3。
concat(sourceFlux, 2)：使用 concat 方法，将这三个 Flux 串联为一个，并设置预取值为 2。这意味着在处理过程中可以提前请求 2 个元素进行处理。
订阅和输出：结果将按顺序输出所有元素，预取会影响内部的调度和流量管理。

prefetch 的作用：

流量管理：通过设置 prefetch，你可以控制一次性从上游请求多少数据，这样在网络通信、远程调用等场景中可以提高吞吐量，避免频繁请求。
平衡性能与资源使用：较大的 prefetch 值可能会提高性能，因为可以减少请求次数，但会占用更多内存来存储预取的数据。较小的 prefetch 值则更节省资源，但可能导致性能下降，因为每次都要等待更多的请求响应周期。

适用场景：

数据流背压处理：如果你需要按顺序处理多个 Publisher，并且每个 Publisher 都发出大量的数据，可以通过设置 prefetch 来优化背压策略，减轻上下游之间的请求响应压力。
大规模异步任务处理：适合用于对大规模异步数据流进行调度和优化，确保流量在可控范围内，并提高整体吞吐量。

扩展用法：

prefetch 参数在流量较大、请求与响应频繁的情况下尤为有用。可以调整 prefetch 的值来观察应用性能的变化，找到合适的参数来平衡吞吐量与资源消耗。

总结：

concat(Publisher<? extends Publisher<? extends T>> sources, int prefetch) 是一个带有 prefetch 参数的 concat 方法，允许你在处理多个 Publisher 时预取一定数量的元素来提高性能。
它适用于需要顺序处理多个数据流且有较大数据量的场景，通过 prefetch 可以优化流量管理和资源使用，特别是在背压管理和异步任务处理时效果明显。

concat

@SafeVarargs
public static <T> Flux<T> concat(Publisher<? extends T>... sources)

Concatenate all sources provided as a vararg, forwarding elements emitted by the sources downstream.

将所有作为可变参数提供的源连接起来，将源发出的元素转发到下游。

Concatenation is achieved by sequentially subscribing to the first source then waiting for it to complete before subscribing to the next, and so on until the last source completes. Any error interrupts the sequence immediately and is forwarded downstream.

Type Parameters:

T - The type of values in both source and output sequences

Parameters:

sources - The Publisher of Publisher to concat

Returns:

a new Flux concatenating all source sequences

类型参数：

T - 源和输出序列中值的类型

参数：

sources - 要连接的发布者的发布者

返回：

一个新的 Flux，连接所有源序列。

concat(Publisher<? extends T>... sources) 是一个 Flux 类的静态方法，用于按顺序将多个 Publisher 实例连接起来。

与其他 concat 方法类似，它会按顺序依次订阅并发射每个 Publisher 中的元素，直到所有 Publisher 都完成。

参数解释：

sources：可变参数列表，表示一组 Publisher，这些 Publisher 将按顺序连接。每个 Publisher 都会发出类型为 T 的元素。

特点：

顺序处理：它会按顺序订阅每个 Publisher，等待一个完成后再订阅下一个。这意味着即使有多个 Publisher 同时发出元素，它们的发射顺序仍然会被严格控制。
按需拉取：与 Reactor 中的其他操作符类似，它在处理流量时也会根据背压策略按需请求元素。
延迟订阅：每当一个 Publisher 完成后，才会开始订阅下一个 Publisher，因此不会同时订阅所有源。

示例代码：

java

import reactor.core.publisher.Flux;

public class ConcatExample {
    public static void main(String[] args) {
        // 创建多个 Flux，每个发出一组字符串
        Flux<String> flux1 = Flux.just("A", "B", "C");
        Flux<String> flux2 = Flux.just("D", "E", "F");
        Flux<String> flux3 = Flux.just("G", "H", "I");

        // 使用 concat 将多个 Flux 按顺序连接
        Flux<String> result = Flux.concat(flux1, flux2, flux3);

        // 订阅并打印每个元素
        result.subscribe(System.out::println);
    }
}

输出结果：

A
B
C
D
E
F
G
H
I

代码详解：

flux1, flux2, flux3：三个不同的 Flux 实例，每个发出一组字符串。
Flux.concat(flux1, flux2, flux3)：使用 concat 方法将这些 Flux 实例连接在一起，它们将按顺序发出元素。即，flux1 发出的元素会首先被处理，接着是 flux2，最后是 flux3。
订阅和输出：订阅 result 后，会按顺序输出每个 Flux 发出的元素。

主要特点：

顺序执行：即使源 Publisher 同时发出数据，concat 仍然会确保它们按顺序发出并且前一个流没有完成时不会订阅下一个流。
支持多个 Publisher：它可以接受多个 Publisher，并按顺序发出它们发射的数据。
背压支持：concat 遵循 Reactive Streams 背压策略，不会一次性请求所有源的数据，而是按需拉取。

适用场景：

顺序处理数据：当你需要确保多个 Publisher 的发射顺序时，例如多个数据库查询结果需要按顺序处理，可以使用 concat。
数据流拼接：适用于需要将多个独立的数据流合并为一个连续流的场景。

扩展用法：

你可以根据需求传入任意多个 Publisher，它们会被顺序处理。
如果某个 Publisher 发送错误，concat 将立即终止，并将该错误信号传递给订阅者。

总结：

@SafeVarargs public static <T> Flux<T> concat(Publisher<? extends T>... sources) 是一种方便的工具，可以将多个 Publisher 按顺序连接起来，确保每个源中的数据按顺序发射。这在需要按顺序处理多个异步数据源时非常有用。

concatDelayError

public static <T> Flux<T> concatDelayError(Publisher<? extends Publisher<? extends T>> sources)

Concatenate all sources emitted as an onNext signal from a parent Publisher, forwarding elements emitted by the sources downstream.

将所有作为父发布者的 onNext 信号发出的源连接起来，将源发出的元素转发到下游。

Concatenation is achieved by sequentially subscribing to the first source then waiting for it to complete before subscribing to the next, and so on until the last source completes. Errors do not interrupt the main sequence but are propagated after the rest of the sources have had a chance to be concatenated.

连接是通过顺序订阅第一个源来实现的，然后等待其完成后再订阅下一个源，以此类推，直到最后一个源完成。错误不会中断主序列，而是在所有源有机会连接后再传播。

Discard Support:

This operator discards elements it internally queued for backpressure upon cancellation.

Type Parameters:

T - The type of values in both source and output sequences

Parameters:

sources - The Publisher of Publisher to concatenate

Returns:

a new Flux concatenating all inner sources sequences, delaying errors

丢弃支持：

该操作符在取消时会丢弃它内部排队的用于背压的元素。

类型参数：

T - 源和输出序列中值的类型

参数：

sources - 要连接的发布者的发布者

返回：

一个新的 Flux，连接所有内部源序列，延迟错误的传播。

concatDelayError(Publisher<? extends Publisher<? extends T>> sources) 是 Reactor Flux 类的一个静态方法，它与 concat 操作符类似，但在处理多个 Publisher 时具有延迟错误处理的功能。

也就是说，即使其中一个 Publisher 发出了错误信号，它仍会继续处理剩下的 Publisher，而不会立即中断整个流的处理。错误会在所有的 Publisher 都完成之后再被发射。

参数解释：

sources：一个 Publisher，它会发出一组 Publisher 实例。这些 Publisher 会按顺序被连接，并发射它们各自的数据。

特点：

延迟错误：当某个 Publisher 发出错误时，错误信号不会立即中断流的处理。concatDelayError 会等待所有的 Publisher 完成后再发射错误信号。这样可以确保即使某些流中出现错误，后续流仍然会被处理。
顺序执行：与 concat 类似，多个 Publisher 按顺序订阅，前一个 Publisher 完成之后才会订阅下一个。
多个 Publisher 的合并：它处理的是一个发出多个 Publisher 的流。

示例代码：

java

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

public class ConcatDelayErrorExample {
    public static void main(String[] args) {
        // 创建几个 Publisher，其中一个会发出错误
        Mono<String> mono1 = Mono.just("A");
        Mono<String> mono2 = Mono.error(new RuntimeException("Error in mono2"));
        Mono<String> mono3 = Mono.just("C");

        // 使用 concatDelayError 将这些 Publisher 合并，即使有错误也会继续处理
        Flux<String> result = Flux.concatDelayError(Flux.just(mono1, mono2, mono3));

        // 订阅并打印每个元素
        result.subscribe(
            System.out::println, 
            error -> System.out.println("Error: " + error.getMessage())
        );
    }
}

输出结果：

A
C
Error: Error in mono2

代码详解：

mono1, mono2, mono3：分别是三个 Mono 实例，mono2 会抛出一个错误，mono1 和 mono3 会正常发出数据。
Flux.concatDelayError(Flux.just(mono1, mono2, mono3))：使用 concatDelayError 将这些 Mono 合并，即使 mono2 抛出错误，mono3 仍然会被处理。
订阅和输出：订阅结果流，输出正常发出的元素，最终输出错误信息。

关键特性：

错误延迟发射：即使中间某个流出错，也不会立即停止处理，而是继续处理其他流。当所有流处理完成后，错误才会被发射。这使得在批处理场景中，多个流的任务可以尽量完成，即使其中一些失败了。
顺序执行：与 concat 一样，它保证按顺序处理每个 Publisher。
多个源的管理：它接收一个发出多个 Publisher 的流，并按顺序发射这些 Publisher 中的数据。

适用场景：

延迟错误场景：适合在多个数据源中有可能发生错误的情况下使用，但希望即使有一个源出错，其他源仍然能够继续执行。
批量任务执行：当你需要执行多个任务，每个任务都可能失败，但希望尽量完成所有任务时，可以使用 concatDelayError。

总结：

public static <T> Flux<T> concatDelayError(Publisher<? extends Publisher<? extends T>> sources) 是一个用于顺序连接多个 Publisher 的方法，同时具有延迟错误发射的特性。即使某个流发生了错误，也会继续处理其他流，并在最后统一发射错误。这对于批处理和延迟错误处理的场景非常实用。

concatDelayError

public static <T> Flux<T> concatDelayError(Publisher<? extends Publisher<? extends T>> sources,
                                           int prefetch)

Concatenate all sources emitted as an onNext signal from a parent Publisher, forwarding elements emitted by the sources downstream.

将所有作为父发布者的 onNext 信号发出的源连接起来，将源发出的元素转发到下游。

Concatenation is achieved by sequentially subscribing to the first source then waiting for it to complete before subscribing to the next, and so on until the last source completes. Errors do not interrupt the main sequence but are propagated after the rest of the sources have had a chance to be concatenated.

连接是通过顺序订阅第一个源来实现的，然后等待其完成后再订阅下一个源，以此类推，直到最后一个源完成。错误不会中断主序列，而是在所有源连接完毕后再进行传播。

Discard Support:

This operator discards elements it internally queued for backpressure upon cancellation.

Type Parameters:

T - The type of values in both source and output sequences

Parameters:

sources - The Publisher of Publisher to concatenate

prefetch - number of elements to prefetch from the source, to be turned into inner Publishers

Returns:

a new Flux concatenating all inner sources sequences until complete or error

丢弃支持：

该操作符在取消时会丢弃它内部排队的用于背压的元素。

类型参数：

T - 源和输出序列中值的类型

参数：

sources - 要连接的发布者的发布者

prefetch - 从源中预取的元素数量，以便转换为内部发布者

返回：

一个新的 Flux，连接所有内部源序列，直到完成或出现错误。

concatDelayError(Publisher<? extends Publisher<? extends T>> sources, int prefetch) 是 Reactor 中 Flux 类的一个静态方法，它和 concatDelayError(Publisher<? extends Publisher<? extends T>> sources) 类似，但增加了一个 prefetch 参数，用来控制每次从上游请求的数据量。

参数说明：

sources：一个 Publisher，它会发出多个 Publisher 实例，这些 Publisher 会按顺序被连接，发出各自的数据。
prefetch：指定预取数据量，即在订阅前从每个 Publisher 请求的元素数量。这个参数决定了 Reactive Streams 背压机制的行为，允许你控制消费速度与生产速度之间的平衡。

特点：

延迟错误处理：即使某个 Publisher 发出错误，错误不会立即中断流的处理，而是等到所有 Publisher 完成后再发出错误。
顺序合并：多个 Publisher 按顺序执行，当前的 Publisher 完成后才会处理下一个。
背压支持：通过 prefetch 参数，可以控制每次从每个源中请求多少数据，从而更好地适应上下游的流速差异。

示例代码：

java

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

public class ConcatDelayErrorExample {
    public static void main(String[] args) {
        // 创建三个 Publisher，其中一个会发出错误
        Mono<String> mono1 = Mono.just("A");
        Mono<String> mono2 = Mono.error(new RuntimeException("Error in mono2"));
        Mono<String> mono3 = Mono.just("C");

        // 使用 concatDelayError 将多个 Publisher 合并，并指定 prefetch 值
        Flux<String> result = Flux.concatDelayError(Flux.just(mono1, mono2, mono3), 1);

        // 订阅并输出结果
        result.subscribe(
            System.out::println,
            error -> System.out.println("Error: " + error.getMessage())
        );
    }
}

输出结果：

A
C
Error: Error in mono2

代码解读：

mono1, mono2, mono3：分别是三个 Mono 实例，其中 mono1 和 mono3 正常发出数据，而 mono2 会抛出错误。
Flux.concatDelayError(Flux.just(mono1, mono2, mono3), 1)：该方法将 mono1、mono2 和 mono3 组合在一起，按顺序处理。prefetch 值为 1，表示每次只请求一个元素。
延迟错误处理：即使 mono2 出错，mono3 仍然会被执行，错误会在 mono3 完成后统一发出。

prefetch 的作用：

控制数据流动：在 Reactive Streams 中，背压是为了防止数据生产过快，而消费过慢。prefetch 参数允许你控制一次性从每个 Publisher 中请求的数据量，从而避免下游被过多的数据淹没。
优化性能：在处理大量数据时，合理设置 prefetch 可以提高吞吐量和性能。在这种情况下，prefetch 可以帮助你根据内存和处理能力来调节数据的流动速度。

使用场景：

延迟错误的批处理场景：适用于需要处理多个数据流，但希望在某些流发生错误时继续处理其他流的场景。比如批处理任务，每个任务独立运行，即使中间有错误，也希望能完成其他任务。
数据流的背压控制：当你需要在处理数据时精确控制消费与生产的节奏时，prefetch 能帮助你通过调节请求的数量来优化处理效率。

总结：

concatDelayError(Publisher<? extends Publisher<? extends T>> sources, int prefetch) 是一种用于合并多个 Publisher 的方法，并通过延迟错误发射确保即使某个流发生错误，其他流仍然可以继续执行。同时，它引入了 prefetch 参数来控制背压，可以在不同的数据流速之间取得平衡，非常适用于需要延迟错误处理的场景，并且对数据流控制有精确需求的场合。

concatDelayError

public static <T> Flux<T> concatDelayError(Publisher<? extends Publisher<? extends T>> sources,
                                           boolean delayUntilEnd,
                                           int prefetch)

Concatenate all sources emitted as an onNext signal from a parent Publisher, forwarding elements emitted by the sources downstream.

将所有作为父发布者的 onNext 信号发出的源连接起来，将源发出的元素转发到下游。

Concatenation is achieved by sequentially subscribing to the first source then waiting for it to complete before subscribing to the next, and so on until the last source completes.

连接是通过顺序订阅第一个源来实现的，然后等待其完成后再订阅下一个源，以此类推，直到最后一个源完成。

Errors do not interrupt the main sequence but are propagated after the current concat backlog if delayUntilEnd is false or after all sources have had a chance to be concatenated if delayUntilEnd is true.

错误不会中断主序列，而是在当前连接的积压结束后传播（如果 delayUntilEnd 为 false），或者在所有源有机会连接后传播（如果 delayUntilEnd 为 true）。

Discard Support:

This operator discards elements it internally queued for backpressure upon cancellation.

Type Parameters:

T - The type of values in both source and output sequences

Parameters:

sources - The Publisher of Publisher to concatenate

delayUntilEnd - delay error until all sources have been consumed instead of after the current source

prefetch - the number of Publishers to prefetch from the outer Publisher

Returns:

a new Flux concatenating all inner sources sequences until complete or error

丢弃支持：

该操作符在取消时会丢弃它内部排队的用于背压的元素。

类型参数：

T - 源和输出序列中值的类型

参数：

sources - 要连接的发布者的发布者

delayUntilEnd - 延迟错误传播，直到所有源都被消费，而不是在当前源后立即传播

prefetch - 从外部发布者预取的发布者数量

返回：

一个新的 Flux，连接所有内部源序列，直到完成或出现错误。

concatDelayError(Publisher<? extends Publisher<? extends T>> sources, boolean delayUntilEnd, int prefetch) 是 Reactor 中 Flux 类的一个静态方法。

这个方法与之前的 concatDelayError 方法类似，不过增加了一个 boolean delayUntilEnd 参数，用来更细粒度地控制错误的延迟处理行为。

参数说明：

sources：这是一个 Publisher，它发出多个 Publisher 实例。这些 Publisher 将按顺序连接，每个 Publisher 发出其数据，然后传递到下一个 Publisher。
delayUntilEnd：此布尔值参数控制是否在所有流完成之后再抛出错误。如果为 true，错误会在所有流完成之后才发出；如果为 false，则只会在当前流完成后抛出错误，并且错误处理会立即开始。
prefetch：控制每次从 sources 中预取的数据量，用于优化 Reactive Streams 背压。

特点：

延迟错误处理：即使某个 Publisher 发出错误，流的处理不会立即中断，而是根据 delayUntilEnd 参数来决定是否继续处理后续的 Publisher。
顺序合并：多个 Publisher 按顺序执行，当前的 Publisher 完成后才会处理下一个。
背压支持：通过 prefetch 参数，可以控制每次从 Publisher 中请求多少数据，提供对数据流的精细控制。

示例代码：

java

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

public class ConcatDelayErrorExample {
    public static void main(String[] args) {
        // 创建三个 Publisher，其中一个会发出错误
        Mono<String> mono1 = Mono.just("A");
        Mono<String> mono2 = Mono.error(new RuntimeException("Error in mono2"));
        Mono<String> mono3 = Mono.just("C");

        // 使用 concatDelayError 将多个 Publisher 合并，并指定 delayUntilEnd 和 prefetch 值
        Flux<String> result = Flux.concatDelayError(Flux.just(mono1, mono2, mono3), true, 1);

        // 订阅并输出结果
        result.subscribe(
            System.out::println,
            error -> System.out.println("Error: " + error.getMessage())
        );
    }
}

输出结果：

A
C
Error: Error in mono2

代码解读：

mono1, mono2, mono3：这三个 Mono 分别发出 "A"、抛出错误、发出 "C"。
Flux.concatDelayError(Flux.just(mono1, mono2, mono3), true, 1)：此方法将 mono1, mono2, mono3 合并，delayUntilEnd 为 true，意味着即使 mono2 出现错误，也不会立即终止流，而是等 mono3 也执行完后再发出错误。prefetch 参数为 1，表示每次从上游请求一个元素。
延迟错误处理：由于 delayUntilEnd 为 true，mono3 在 mono2 出错后仍然可以正常发出数据，错误将在所有流结束后抛出。

delayUntilEnd 参数的作用：

true：如果设置为 true，错误会在所有流都完成之后再抛出，保证所有的流都执行完毕。这种行为适用于批处理任务，即使某个任务失败，也希望其他任务继续执行，最后再统一报告错误。
false：如果设置为 false，一旦某个流发生错误，当前流会处理完并抛出错误，后续的流将被跳过。这适用于当某个流发生错误时，希望立即处理错误并停止后续任务的场景。

prefetch 的作用：

prefetch 控制每次从 sources 请求的数据量。在 Reactive Streams 中，控制数据流动速率可以防止过多数据涌入下游，从而避免内存和处理性能问题。通过设置合理的 prefetch，你可以确保系统以稳定的速度消费数据。

使用场景：

容错性任务处理：如果你有多个任务需要串行执行，并且希望即使中间有任务失败，也希望其他任务继续执行，delayUntilEnd = true 可以确保任务的容错性。
数据流控制：当你需要在处理多个数据流时精细控制其速率，prefetch 参数可以帮助优化系统的吞吐量和性能。

总结：

concatDelayError(Publisher<? extends Publisher<? extends T>> sources, boolean delayUntilEnd, int prefetch) 是一个用于合并多个 Publisher 的方法，通过 delayUntilEnd 控制错误的处理时机，并通过 prefetch 控制数据的流动速率。这个方法非常适合需要延迟错误处理、并行任务处理以及精细控制背压的场景。

concatDelayError

@SafeVarargs
public static <T> Flux<T> concatDelayError(Publisher<? extends T>... sources)

Concatenate all sources provided as a vararg, forwarding elements emitted by the sources downstream.

将所有作为可变参数提供的源连接起来，将源发出的元素转发到下游。

Concatenation is achieved by sequentially subscribing to the first source then waiting for it to complete before subscribing to the next, and so on until the last source completes. Errors do not interrupt the main sequence but are propagated after the rest of the sources have had a chance to be concatenated.

连接是通过顺序订阅第一个源来实现的，然后等待其完成后再订阅下一个源，以此类推，直到最后一个源完成。错误不会中断主序列，而是在所有源有机会连接后再进行传播。

Discard Support:

This operator discards elements it internally queued for backpressure upon cancellation.

Type Parameters:

T - The type of values in both source and output sequences

Parameters:

sources - The Publisher of Publisher to concat

Returns:

a new Flux concatenating all source sequences

丢弃支持：

该操作符在取消时会丢弃它内部排队的用于背压的元素。

类型参数：

T - 源和输出序列中值的类型

参数：

sources - 要连接的发布者的发布者

返回：

一个新的 Flux，连接所有源序列。

@SafeVarargs public static <T> Flux<T> concatDelayError(Publisher<? extends T>... sources) 是一个使用可变参数的静态方法，提供了在多个 Publisher 之间进行延迟错误处理的功能。

与其他 concatDelayError 方法类似，这个方法将多个 Publisher 顺序合并成一个 Flux，并且在其中一个 Publisher 出现错误时不会立刻中断处理，而是等所有 Publisher 完成后再抛出错误。

参数说明：

sources：这是可变参数形式的 Publisher，代表多个 Publisher 对象，这些 Publisher 会被按顺序进行连接，形成一个连续的数据流。
可变参数 (...)：可以传递多个 Publisher 对象，而不需要手动将它们封装成数组，这使得方法调用更加简洁。

特点：

延迟错误处理：即使某个 Publisher 发出错误，流不会立即中断，直到所有 Publisher 完成才会处理并抛出错误。
顺序执行：多个 Publisher 将按顺序执行，当前 Publisher 完成后，才会开始下一个 Publisher。
安全使用变参：由于这个方法使用了 @SafeVarargs 注解，保证了在使用可变参数时的类型安全。

示例代码：

java

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

public class ConcatDelayErrorExample {
    public static void main(String[] args) {
        // 创建多个 Publisher，其中一个会发出错误
        Mono<String> mono1 = Mono.just("A");
        Mono<String> mono2 = Mono.error(new RuntimeException("Error in mono2"));
        Mono<String> mono3 = Mono.just("C");

        // 使用 concatDelayError 将多个 Publisher 合并
        Flux<String> result = Flux.concatDelayError(mono1, mono2, mono3);

        // 订阅并输出结果
        result.subscribe(
            System.out::println,
            error -> System.out.println("Error: " + error.getMessage())
        );
    }
}

输出结果：

A
C
Error: Error in mono2

代码解读：

mono1, mono2, mono3：这是三个 Mono 对象，分别发出 "A"，抛出错误，和发出 "C"。
Flux.concatDelayError(mono1, mono2, mono3)：这个方法将 mono1, mono2, mono3 合并成一个 Flux，即使 mono2 抛出了错误，由于这是延迟错误处理，流不会立即终止，mono3 仍然可以发出它的值。
错误延迟抛出：mono2 虽然发生了错误，但 mono3 依然能够执行，等 mono3 处理完后，错误才会被抛出。