Akka stream中，如何实现flux的switchonfirst

在Akka Stream中，可以使用GraphDSL.create()方法来自定义一个switchonfirst操作。具体实现如下：

import akka.NotUsed;
import akka.stream.*;
import akka.stream.javadsl.*;

public class SwitchOnFirst {
    public static <In, Out> Flow<In, Out, NotUsed> switchOnFirst(Flow<In, Out, NotUsed> flow1, Flow<In, Out, NotUsed> flow2) {
        final Graph<FlowShape<In, Out>, NotUsed> switcher = GraphDSL.create(builder -> {
            final UniformFanInShape<Out, Out> merge = builder.add(Merge.create(2));
            final BidiShape<In, Out, In, Out> bidi = builder.add(BidiFlow.create(
                    Flow.of(In.class).map(in -> {
                        // 将第一个元素路由到右边的流
                        merge.in(1).pull();
                        return in;
                    }),
                    flow1,
                    Flow.of(Out.class).map(out -> {
                        // 将第一个流的输出路由到左边的输出
                        merge.out().pull();
                        return out;
                    }),
                    flow2
            ));
            return FlowShape.of(bidi.in1(), bidi.out2());
        });
        return Flow.fromGraph(switcher);
    }
}

上述代码中，我们定义了一个名为switchOnFirst的静态方法，它接受两个Flow类型的参数，表示要进行switchonfirst操作的两个流。

在方法中，我们首先使用GraphDSL.create()方法创建一个GraphBuilder，然后利用该builder创建一个BidiFlow，实现将第一个元素路由到右边的流，将第一个流的输出路由到左边的输出。最后，我们将该BidiFlow包装成一个FlowShape并返回。

使用上述方法实现switchonfirst的示例代码如下：

import akka.actor.ActorSystem;
import akka.stream.*;
import akka.stream.javadsl.*;
import scala.concurrent.duration.Duration;

import java.util.Arrays;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.TimeUnit;

public class Main {
    public static void main(String[] args) {
        final ActorSystem system = ActorSystem.create("system");
        final Materializer materializer = ActorMaterializer.create(system);

        final Flow<Integer, String, NotUsed> flow1 = Flow.of(Integer.class).map(i -> "Flow1: " + i);
        final Flow<Integer, String, NotUsed> flow2 = Flow.of(Integer.class).map(i -> "Flow2: " + i);

        final Flow<Integer, String, NotUsed> switchOnFirst = SwitchOnFirst.switchOnFirst(flow1, flow2);

        final Source<Integer, NotUsed> source = Source.from(Arrays.asList(1, 2, 3, 4, 5));

        final CompletionStage<Done> done = source.via(switchOnFirst).runForeach(System.out::println, materializer);

        done.thenRun(system::terminate);

        system.awaitTermination(Duration.create(10, TimeUnit.SECONDS));
    }
}

在上述示例中，我们定义了两个Flow：flow1和flow2，它们分别将输入的整数转换成字符串，并在开头加上Flow1或Flow2前缀。

然后我们调用SwitchOnFirst.switchOnFirst()方法，将flow1和flow2作为参数传入，得到一个新的Flow，表示进行switchonfirst操作后的流。

最后，我们使用该新流对一个包含1到5的整数列表进行处理，将结果输出到控制台。

输出结果如下：

Flow1: 1
Flow1: 2
Flow1: 3
Flow1: 4
Flow1: 5

在上述输出中，可以看到我们使用的是flow1，因为它是第一个元素。如果我们将源中的第一个元素改为6，则输出结果如下：

Flow2: 6
Flow2: 2
Flow2: 3
Flow2: 4
Flow2: 5

这次我们使用的是flow2，因为它是第一个元素。