java - 如何在响应式(Reactive) Java 中将新对象添加到现有流中？

Question

假设我已经有一个 react 流，现在我想向这个现有流中再添加一个对象。我该怎么做？

这是 approach我发现，这是要走的路吗？

import java.util.ArrayList;
import java.util.List;
import java.util.function.Consumer;

import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;

/**
 * Created by ton on 10/11/16.
 */
public class Example {

    private List<FluxSink<String>> handlers = new ArrayList<>();

    public Flux<String> getMessagesAsStream() {
        Flux<String> result = Flux.create(sink -> {
                handlers.add(sink);
            sink.setCancellation(() -> handlers.remove(sink));
        });

        return result;
    }

    public void handleMessage(String message) {
        handlers.forEach(han -> han.next(message));
    }

    public static void main(String[] args) {
        Example example = new Example();
        example.getMessagesAsStream().subscribe(req -> System.out.println("req = " + req));
        example.getMessagesAsStream().subscribe(msg -> System.out.println(msg.toUpperCase()));
        example.handleMessage("een");
        example.handleMessage("twee");
        example.handleMessage("drie");
    }
}

Answer 1

假设这是您现有的流:

Flux<Integer> existingStream = Flux.just(1, 2, 3, 4);

您可以连接两个流:

Flux<Integer> appendObjectToStream = Flux.concat(existingStream, Flux.just(5));

这将产生 [1, 2, 3, 4, 5]。

或者，您可以合并两个流:

Flux<Integer> mergeObjectWithStream = Flux.merge(existingStream, Flux.just(5));

这将产生类似的流，但是 5 元素可能出现在生成的通量中的任何位置。