Spring Webflux型合并流量的安全性

Question

在服务器上，我添加了一个字符串类型的流量连接：

Flux<String> addDataConnection() {
        return Flux.create(emitter -> {
            fluxSinks.add(emitter);
        });
    }

然后，每当我收到消息时，它就会传播消息：

void propagateMessage(String messageContent) {
        fluxSinks.parallelStream()
                .forEach(fluxSink -> fluxSink.next(messageContent));
    }

这些数据来自外部来源，因此传入事件的频率超出了我的控制范围。

Flux<String> addHeartbeat() {
    return Flux.interval(Duration.ofSeconds(2))
            .map(i -> "heartbeat");
}

Flux<String> initiateFluxStreams() {
        return Flux.merge(addDataConnection(), addHeartbeat());
    }

Answer 1

对于不同类型的合并通量，没有什么“神奇”的方法可以做到这一点，因为您不能保证这些类型将有任何共同的祖先（除了Object，但正如您所指出的，这并不太有用）

如果您的类型是foo，则可以使用foo的“特殊”实例（或子类）来表示空的heartbeat类型，但我并不特别喜欢这种方法--它可能会做出设计牺牲，以便在其中插入某个类型。

我首选的方法是使用流量 > optional，其中的“heartbeat”触发器只是空选项。类似于：

Flux<Foo> addDataConnection() {
    return Flux.create(emitter -> {
        fluxSinks.add(emitter);
    });
}

Flux<Optional<Foo>> initiateFluxStreams() {
    return Flux.merge(addDataConnection().map(f -> Optional.of(f)), addHeartbeat().map(h -> Optional.empty()));
}

Flux<Long> addHeartbeat() {
    return Flux.interval(Duration.ofSeconds(2));
}

然后，您只需筛选并映射结果流量，以便在处理之前清除心跳：

fluxOfOptionalFoo
        .filter(Optional::isPresent)
        .map(Optional::get);

我更喜欢这种方法，原因有很多：

• 不需要对基础对象结构进行任何更改
• 更具描述性，您有一个流量类型“可能在那里，也可能不在那里”
• 容易过滤掉空选项（如果您想用heartbeat函数做一些特别的事情，则不需要）