Эффективно присоединяйтесь к сортированным производителям флюса ⇐ JAVA

[Anonymous]

Мне нужно объединить большие потоки данных, доступные через Flux, каждый из которых содержит метку времени и значение. Если временные метки совпадают, то значения необходимо просуммировать. Данные в потоке сортируются по метке времени в порядке возрастания.
Для небольших потоков я бы использовал функцию groupBy, но поскольку поток содержит много записей, это не так. эффективный.
Я хотел бы воспользоваться тем фактом, что записи в потоке упорядочены, но не могу найти правильную конструкцию. Каковы инструменты для достижения чего-то подобного? Ниже приведен код sudo того, что я хочу сделать:
var flux1 = Flux.just(
new Data(ZonedDateTime.parse("2025-01-01T00:00:00"), 1.0),
new Data(ZonedDateTime.parse("2025-03-01T00:00:00"), 1.0)
);

var flux2 = Flux.just(
new Data(ZonedDateTime.parse("2025-02-01T00:00:00"), 2.0),
new Data(ZonedDateTime.parse("2025-03-01T00:00:00"), 2.0),
new Data(ZonedDateTime.parse("2025-04-01T00:00:00"), 2.0)
);

var flux3 = Flux.just(
new Data(ZonedDateTime.parse("2025-02-01T00:00:00"), 5.0)
);

var input = List.of(flux1, flux2, flux3);

var output = Flux.create(sink -> {
List nextEntries = input.stream().map(Flux::next).toList();

do {
ZonedDateTime nextTimestamp = nextEntries.stream().map(Data::getTimestamp).min(ZonedDateTime::compareTo).get();
List affectedStreams = IntStream.range(0, input.size()).filter(i -> nextTimestamp == nextEntries.getTimestamp()).toList();
double nextOutput = affectedStreams.stream().mapToDouble(i -> nextEntries.getValue()).sum();
sink.next(new Data(nextTimestamp, nextOutput));
affectedStreams.forEach(i -> nextEntries = input.get(i).next());
} while (!allFluxAreConsumed);
});

// expected output:
// [
// Data(ZonedDateTime.parse("2025-01-01T00:00:00"), 1.0),
// Data(ZonedDateTime.parse("2025-02-01T00:00:00"), 7.0),
// Data(ZonedDateTime.parse("2025-03-01T00:00:00"), 3.0),
// Data(ZonedDateTime.parse("2025-05-01T00:00:00"), 2.0)
// ]


Подробнее здесь: https://stackoverflow.com/questions/793 ... fficiently