合并热通量源

Question

在带有Reactor的Spring Boot 2中，我试图合并两个Flux热源。但是，merge似乎只报告了Flux中两个merge参数中的第一个。如何让merge识别第二个Flux。

在下面的例子中，当System.err是第一个参数时，B-2中的outgoing1a甚至不打印。如果我首先制作outgoing2，那么A-2不会打印。

以下是完整的例子;

package com.example.demo;

import java.time.Duration;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

public class Weather {
String city;
Integer temperature;

public Weather(String city, Integer temperature) {
    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<>();
    BlockingQueue<Weather> queue2 = new LinkedBlockingQueue<>();

    // Assume Spring @Repository "A-1"
    new Thread(() -> {
        for (int d = 1; d < 1000; 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(); 

    // Assume Spring @Repository "B-1"
    new Thread(() -> {
        for (int d = 1; d < 1000; d += 1) {
            for (String s: new String[] {"MOS", "TLV"}) {
                queue2.add(new Weather(s, d));
                try { Thread.sleep(1000); } catch (InterruptedException e) {}
            }
        }
    }).start();

    // Assume Spring @Service "A-2" = real-time LDN, NYC, PAR, ZUR
    Flux<Weather> outgoing1 = Flux.<Weather>create(
        sink -> {
            for (int i = 0; i < 1000; i++) {
                try {
                    sink.next(queue.take());
                    System.err.println("1 " + queue.size());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            sink.complete();
        }
    ).publishOn(Schedulers.newSingle("outgoing-1"));

    // Assume Spring @Service "B-2" = real-time MOS, TLV
    Flux<Weather> outgoing2 = Flux.<Weather>create(
            sink -> {
                for (int i = 0; i < 1000; i++) {
                    try {
                        sink.next(queue2.take());
                        System.err.println("2 " + queue2.size());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                sink.complete();
            }
        ).publishOn(Schedulers.newSingle("outgoing-2"));

    // Assume Spring @Service "A-3" = 5 second summary of LDN, NYC, PAR, ZUR
    Flux<Weather> outgoing1a = Flux.from(outgoing1)   
        .groupBy(c -> c.city)
        .flatMap(g -> g
            .sample(Duration.ofSeconds(5))
        )
        .log("C");

    // Assume Spring @Service "C" - merges "A-3" and "B-2"
    // only prints outgoing1a
    Flux.merge(outgoing1a, outgoing2).subscribe(System.out::println); 

    // only prints outgoing2
    //Flux.merge(outgoing2, outgoing1a).subscribe(System.out::println); 

}
}

Answer 1

这里有一些事情在起作用。

1. 请注意.merge运营商的以下建议......

请注意，合并是为了适应异步源或有限源而定制的。处理尚未在专用调度程序上发布的无限源时，必须在其自己的调度程序中隔离该源，否则合并会在订阅其他源之前尝试将其消耗掉。

2. 您的出站通量使用.publishOn，但这仅影响在.publishOn运算符之后链接的运算符。即它不会影响.publishOn之前的任何事情。具体来说，它不会影响lambda传递给Flux.create的代码执行的线程。如果你在每个出站通量中的.log()之前添加.publishOn，你可以看到这个。
3. 传递给Flux.create的lambda称为阻塞方法（queue.take）。

因为你在subscribe(...)线程中对合并的Flux调用main，所以传递给Flux.create的lambda在main线程中执行，并阻塞它。

最简单的解决方法是使用.subscribeOn而不是.publishOn，以便传递给Flux.create的lambda中的代码在不同的线程（main除外）上运行。这将阻止main线程阻塞，并允许两个出站流的合并输出交错。