过滤 Flux 反应器中的重复对象

Question

如何过滤我自己的重复对象 class？例如：我有一个 class 具有以下属性：

public class MyClass{
  private String id;
  private Date lastUpdate;
  //Getter and Setter are available
}

现在我有一个包含很多 MyClass 对象的 Flux。它们可以包含相同的 ID 但具有不同的日期。我想删除所有具有相同 ID 的对象，但我想保留最后更新的对象。

Flux<MyClass> fluxMyContainer = Flux.just(new MyClass("1", "2017-01-02), 
  new MyClass("2", "2018-11-05"), new MyClass("1", "2018-05-04"));

在这种情况下，我想删除第一个元素。有谁知道如何在不阻塞的情况下做到这一点？

Answer 1

您可以：

groupBy id
reduce 组丢弃旧条目
flatMap Monos

Flux.just(...)
    .groupBy(MyClass::getId)
    .flatMap(g -> g.reduce((a, b) -> a.getLastUpdate().compareTo(b.getLastUpdate()) > 0 ? a : b))
    .subscribe(System.out::println);

例如：

Flux.just(
    new MyClass("1", LocalDate.of(1990, 2, 3)),
    new MyClass("1", LocalDate.of(2000, 3, 1)),
    new MyClass("2", LocalDate.of(1000, 2, 2)),
    new MyClass("2", LocalDate.of(2010, 4, 5)),
    new MyClass("2", LocalDate.MIN),
    new MyClass("3", LocalDate.MIN))
     .groupBy(MyClass::getId)
    .flatMap(g -> g.reduce((a, b) -> a.getLastUpdate().compareTo(b.getLastUpdate()) > 0 ? a : b))
    .subscribe(System.out::println);

打印：

MyClass{id='1', lastUpdate=2000-03-01}
MyClass{id='2', lastUpdate=2010-04-05}
MyClass{id='3', lastUpdate=-999999999-01-01}

Answer 2

在 reactor 中删除重复项会阻塞整个管道，没有 non-blocking 方法，无论您使用哪种解决方案

因为删除重复逻辑本身需要在发出结果之前遍历所有项目。其他需要首先遍历所有项目的逻辑（如排序通量）也会阻塞管道，如果你想归档 non-blocking 管道，请谨慎使用此类逻辑。

不知道你为什么会出现这种情况，如果你坚持a)不需要重复和b)non-blocking，那就改变你的设计。

例如，过滤后的结果（唯一的）需要在UI中显示，那么你可以先在UI中一个一个的展示原始数据。如果来自 flux 的新发射项目是重复的，只需更新 UI.

中已经显示的项目

Answer 3

您好，如果您只想过滤重复项，请使用 .distinc()，https://projectreactor.io/docs/core/release/api/reactor/core/publisher/Flux.html#distinct-

  `Flux.just(
    new MyClass("1", LocalDate.of(1990, 2, 3)),
    new MyClass("1", LocalDate.of(2000, 3, 1)),
    new MyClass("2", LocalDate.of(1000, 2, 2)),
    new MyClass("2", LocalDate.of(2010, 4, 5)),
    new MyClass("2", LocalDate.MIN),
    new MyClass("3", LocalDate.MIN))
   .distinct(MyClass::getId)
   .subscribe(System.out::println);`

结果可以。

   `MyClass{id='1', lastUpdate=2000-03-01}
    MyClass{id='2', lastUpdate=2010-04-05}
    MyClass{id='3', lastUpdate=-999999999-01-01}`

过滤 Flux 反应器中的重复对象

Filter duplicate object in Flux reactor

spring

project-reactor

spring-webflux