我之前将实体保存到表中，如下所示：
insertTableOne(foo).then(insertTableTwo(foo)).then(Mono.just(foo));
团队成员建议使用以下方法.and()来并行化插入件：
insertTableOne(foo).and(insertTableTwo(foo).thenRetuen(foo);
虽然我同意它看起来更简洁，但我找不到任何文档表明使用.and()将同时执行Schedulers.parallel()两个插入。我认为它仍然是连续的，除非明确说明与并行。
我错了吗？
我目前正在编写一个方法，该方法需要在继续下游操作之前将每个元素添加到并发HashMap中。例如
public Flux<Bar> methodOne(final Flux<Foo> foos) {
Map<UUID, Foo> idToFoo = new ConcurrentHashMap<>();
return foos.doOnNext(foo -> idToFoo.put(foo.getId(), foo)
.flatMap(foo -> fooToBar(foo, idToFoo));
}
我是否可以确定fooToBar()传递的地图中将填充相应的元素？
据我所知，doOnNext()它是同步的，它会在继续下游之前执行并完成 lambda。然而，我遭到了一位高级开发人员的反对 - 他们认为这是一种副作用，是一种“发射后不管”的类型。
我正在努力更好地理解源代码，以便能够证明用例的合理性，但也想在这里检查。
谢谢
我有类似这样的实用方法。
public static WebClient.ResponseSpec retrieve(final String baseUrl, final Duration responseTimeout) {
// ...
}
public static <T> Flux<T> retrieveBodyToFlux(final String baseUrl, final Duration responseTimeout,
final Class<T> elementClass) {
return retrieve(baseUrl, responseTimeout)
.bodyToFlux(elementClass);
}
public static Mono<Void> download(final String baseUrl, final Duration responseTimeout,
final Path destination, final OpenOption... openOptions) {
return DataBufferUtils.write(
retrieveBodyToFlux(baseUrl, responseTimeout, DataBuffer.class),
destination,
openOptions
);
}
现在我想添加另一种方法让客户端使用临时文件。
// Let me download the file, and you can just cononsume the file!
public static Mono<Void> download(final String baseUrl, final Duration responseTimeout,
final Consumer<? super Path> consumer) {
// Create a temp file
// download the URL to the file
// accept the file to the consumer; possibly blocking IO operations
// don't forget to delete the file!
}
我怎样才能在没有任何阻塞呼叫警告的情况下完成这项工作？
我试过了（似乎有效），但我不知道我做了什么。这是最佳选择吗？我还有其他方法吗？
return Mono.usingWhen(
Mono.fromCallable(() -> Files.createTempFile(null, null)).subscribeOn(Schedulers.boundedElastic()),
p -> download(baseUrl, responseTimeout, p, StandardOpenOption.WRITE)
.then(Mono.just(p))
.doOnNext(consumer).subscribeOn(Schedulers.boundedElastic()),
p -> Mono.fromCallable(() -> {
Files.delete(p);
return null;
}).publishOn(Schedulers.boundedElastic()).then()
)
.then();
我找不到使用 Mono.block() 和 Mono.subscribe() 的区别
对我来说，当使用这两种方法时，代码的行为完全相同。但它不应该。
对于 Mono.block() 我的期望是调用它的线程将阻塞并等待结果，但它在 Mono 的 map 方法中使用，并且基本上会自行解除阻塞。
我有以下使用 Mono.block() 的代码片段：
void doBlocking() {
final var myMono = Mono.just("test").map(elem -> {
System.out.printf("On thread: [%s] inside map\n",Thread.currentThread().getName());
return elem;
});
String value;
System.out.printf("On thread: [%s] before block\n",Thread.currentThread().getName());
value = myMono.block();
System.out.printf("On thread: [%s] after block\n",Thread.currentThread().getName());
System.out.println(value);
}
当我调用此代码时，我收到以下内容：
On thread: [main] before block
On thread: [main] inside map
On thread: [main] after block
test
根据我的理解 Mono.block() 是阻塞方法，所以我假设线程将像获取锁时一样被阻塞。相反，线程用于在Mono 的映射内部执行代码，这意味着它根本不会被阻塞。
对于Mono.subscribe()我希望调用 subscribe 的线程将继续而不等待结果，但它的行为与使用Mono.block()时完全相同
我有一个类似的片段，但现在使用订阅而不是块
void doSubscribing() {
final var myMono = Mono.just("test").map(elem -> {
System.out.printf("On thread: [%s] inside map\n",Thread.currentThread().getName());
return elem;
});
AtomicReference<String> value = new AtomicReference<>();
System.out.printf("On thread: [%s] before subscribe\n",Thread.currentThread().getName());
myMono.subscribe(value::set);
System.out.printf("On thread: [%s] after subscribe\n",Thread.currentThread().getName());
System.out.println(value);
}
当我再次调用此代码时，我得到相同的结果：
On thread: [main] before subscribe
On thread: [main] inside map
On thread: [main] after subscribe
test
我希望当我调用 subscribe 时，当前线程将继续工作，可能显示：
On thread: [main] after subscribe
null
就我而言，阻止和订阅的行为完全相同，那么真正的区别是什么？
A用于获取源码。在下面的函数中，如果拥有map源，则源是一个，否则源是一个。函数返回基于A的List可以查询的所有源，列表大小可以是1。问题是当Stream.map输出可以是两者之一时，如何处理这个Mono和Flux组合成Flux。FluxaMono
class A {
String x;
String y;
boolean isOwner();
}
Flux<Source> foo(List<A> a) {
Flux<Source> sources = a.stream().map(v -> if (v.isOwner) {
Flux<Source> ownedSource = ...; return ownedSource;
} else {
Mono<Source> givenSource = ...; return givenSource;
}
}.???
return sources;
}