java - 如何在热源上使用 groupBy

Question

给出以下代码；

我有一个假的“热源”，我想在其中每 2 秒打印每个城市的最后一个值。我看到 log 点 A 和 B 的行为符合我的预期。但是，代码在 groupBy 上阻塞，并且只有 every 在 log 点 C 处发出最终值。我怎样才能让“C”每 2 秒发出一次。

public class Weather {
    String city;
    Integer temperature;

public Weather(String city, Integer temperature) {
    super();
    this.city = city;
    this.temperature = temperature;
}

@Override
public String toString() {
    return "Weather [city=" + city + ", temperature=" + temperature + "]";
}

public static void main(String[] args) {

    BlockingQueue<Weather> queue = new LinkedBlockingQueue<>();

    new Thread(() -> {
        for (int d = 1; d < 100; d += 1) {
            for (String s: new String[] {"LDN", "NYC", "PAR", "ZUR"}) {
                queue.add(new Weather(s, d));
                try { Thread.sleep(250); } catch (InterruptedException e) {}
            }
        }
    }).start(); 

    Flux<Weather> outgoing = Flux.create(
        sink -> {
            for (int i = 0; i < 100; i++) {
                try {
                    sink.next(queue.take());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            sink.complete();
        }
    );

    ConnectableFlux<Weather> subscriber = outgoing.publish();
    subscriber
    .buffer(Duration.ofSeconds(2))
    .log("A")
    .flatMap(Flux::fromIterable)
    .log("B")
    .groupBy(c -> c.city)
    .flatMap(Flux::last)
    .log("C")           

    .subscribe(s -> System.out.println(">>>>>" + s));


    subscriber.connect();
    System.exit(0);
}

}

Answer 1

这似乎有效；

        subscriber
          .groupBy(c -> c.city)
          .flatMap(g -> g
            .take(Duration.ofSeconds(5))
            .takeLast(1)
          )
          .subscribe(s -> System.out.println(">>>>>" + s));