Mono\Flux上map和subscribe的区别？
Mono 和 Flux 的 map 和 subscribe 区别
基础概念
Mono 和 Flux 是 Project Reactor 中的两个核心类，用于处理异步数据流。Mono 表示一个单一的数据元素，而 Flux 表示多个数据元素的流。
- Mono：表示一个单一的、可能是延迟的结果。
- Flux：表示一个包含 0 到 N 个元素的异步序列。
map 操作
map 是一个操作符，用于对数据流中的每个元素进行转换。它接受一个函数作为参数，这个函数会被应用到每个元素上，生成一个新的元素。
Flux<Integer> flux = Flux.just(1, 2, 3);
Flux<String> mappedFlux = flux.map(i -> "Value: " + i);在这个例子中，map 将每个整数转换为字符串。
subscribe 操作
subscribe 是一个终端操作，用于启动数据流的消费过程。它接受一个或多个订阅者（Subscriber），这些订阅者会处理数据流中的元素。
Flux<Integer> flux = Flux.just(1, 2, 3);
flux.subscribe(
i -> System.out.println("Received: " + i),
err -> System.err.println("Error: " + err),
() -> System.out.println("Done")
);在这个例子中，subscribe 启动了数据流的处理，并定义了如何处理每个元素、错误和完成事件。
区别
- 目的：
- map：用于对数据流中的每个元素进行转换，生成一个新的数据流。
- subscribe：用于启动数据流的消费过程，定义如何处理数据流中的元素。
- 返回值：
- map：返回一个新的
Flux或Mono，表示转换后的数据流。 - subscribe：不返回任何值，它只是启动数据流的处理。
- map：返回一个新的
- 链式操作：
- map：可以与其他操作符（如
filter、flatMap）链式调用，形成复杂的数据处理流程。 - subscribe：是终端操作，不能与其他操作符链式调用。
- map：可以与其他操作符（如
应用场景
- map：当你需要对数据流中的每个元素进行某种转换（如类型转换、数据格式化）时，使用
map。 - subscribe：当你准备好消费数据流并处理其中的元素时，使用
subscribe。
示例代码
import reactor.core.publisher.Flux;
public class MapVsSubscribeExample {
public static void main(String[] args) {
Flux<Integer> flux = Flux.just(1, 2, 3);
// 使用 map 进行转换
Flux<String> mappedFlux = flux.map(i -> "Value: " + i);
// 使用 subscribe 消费数据流
mappedFlux.subscribe(
s -> System.out.println("Received: " + s),
err -> System.err.println("Error: " + err),
() -> System.out.println("Done")
);
}
}参考链接
通过以上解释和示例代码，你应该能够理解 Mono 和 Flux 中 map 和 subscribe 的区别及其应用场景。
相关·内容
我有以下几点：
Flux<String> flux = ...
Mono<Void> mono = ...
Mono<Void> combined = operation(flux, mono);
它们代表并行发生的操作。
现在，我想将flux发出的所有元素输出到sysout，直到mono完成，这里使用的正确运算符是什么？
我试过了：
final Disposable subscribe = flux.subscribe(System.out::println);
mono.doOnSuccessOrError((o, e) -> subscrib
浏览 1提问于2019-06-11得票数 0
3回答
我有一个Flux和Mono，我不确定如何组合它们，以便在Flux的每一项中都有mono值。
我正在尝试这种方法，但它不起作用：
Mono<String> mono1 = Mono.just("x");
Flux<String> flux1 = Flux.just("{1}", "{2}", "{3}", "{4}");
Flux.zip(mono1, flux1, (itemMono1, itemFlux1) -> "-[" + itemFlux1 + itemMo
浏览 20提问于2018-07-31得票数 8
回答已采纳
我有一个用户名列表，希望在不阻塞主线程的情况下从远程服务获取用户详细信息。我使用的是Spring的反应式客户端WebClient。对于响应，我获取Mono，然后订阅它并打印结果。 private Mono<User> getUser(String username) {
return webClient
.get()
.uri(uri + "/users/" + username)
.retrieve()
.bodyToMono(User.class)
浏览 87提问于2019-12-04得票数 3
回答已采纳
关于webflux reactive存储库的小问题，特别是关于Flux saveAll(Iterable Var1)和Flux saveAll(Publisher Var1)的方法； 为了进行比较，我写了以下内容： @Controller
@SpringBootApplication
public class Application {
public static void main(String[] args) {
SpringApplication.run(Application.class);
}
@Autowired
private
浏览 119提问于2020-09-01得票数 1
回答已采纳
我对Mono和Flux很陌生。我正在尝试加入几个下游API响应。这是一个传统的阻塞应用程序。我不想收集Mono的列表，我需要从下游API返回的有效负载列表，这是我从Mono获取的。但是，有时返回给控制器的“结果”只有部分或没有下游API响应。正确的方法是什么？我读过几篇的文章
您不应该在web应用程序中的任何地方调用订阅。如果这是绑定到HTTP请求，则基本上是在触发反应性管道，而不保证资源或完成。调用订阅会触发管道，但不会等到管道完成后再调用
我应该使用CompletableFuture吗？
在我的服务中，我试图
var result = new ArrayList<List&l
浏览 36提问于2022-06-04得票数 2
回答已采纳
2回答
根据Mono和Flux的定义，它们都表示异步数据序列，在订阅之前不会发生任何事情。
出版商有两大家族:热的和冷的。Mono和Flux为每个订阅重新生成数据。如果没有创建订阅，则永远不会生成数据。
另一方面，热门出版商不依赖于任何数量的订户。
这是我的冷流代码：
System.out.println("*********Calling coldStream************");
Flux<String> source = Flux.fromIterable(Arrays.asList("ram", "sa
浏览 7提问于2018-01-03得票数 5
回答已采纳
我对反应性流很陌生，我学习使用concat/串联using方法将两个发布者(Flux是特定的)结合在一起。
我可以用concat方法做的所有事情，同样可以使用concatWith方法来实现。下面是我使用的示例案例：
Mono<String> mono1 = Mono.just(" karan ");
Mono<String> mono2 = Mono.just(" | verma ");
Mono<String> mono3 = Mono.just(" | kv "
浏览 2提问于2018-01-02得票数 2
回答已采纳
1回答
我是第一次接触Project Reactor和R2DBC。如何使用Spring Data R2DBC反应式存储库正确地接收和合并Flux<Child>和Mono<Parent>？ 父级： @Table("parent")
public class Parent{
@Id
private Long id;
private String name;
@Transient
private Flux<Child> children; ParentRepository： interface ParentRepository extends Rea
浏览 20提问于2020-01-21得票数 4
回答已采纳
2回答
我需要为一个函数返回Mono / Flux，但这有2个嵌套订阅。我正在寻找一个更好的解决方案，只有在这两个订阅值可用后才发布Mono/Flux，然后执行一些操作来解除对finalValue的信任。
最终目标是，函数getFinalValue()的订阅者应该能够订阅最终值。我对Flux也有类似的需求。实现这一目标的最佳方法应该是什么？
fun <T> getFinalValue(): Mono<T> {
object1.getValue1().subscribe { value1 ->
object2.getValue2(value1.i
浏览 51提问于2020-05-09得票数 0
回答已采纳
我想，即使在多次阅读javadocs之后，除了同步和异步的flatMap之外，我也找不到map和transforms.In之间的区别，下面的代码我没有看到任何事件(它的行为就好像没有订阅()一样)。
final Flux<GroupedFlux<String, TData>> groupedFlux =
flux.groupBy(Event::getPartitionKey);
groupedFlux.subscribe(g -> g.delayElements(Duration.ofMillis(100))
浏览 3提问于2021-10-29得票数 2
回答已采纳
我有下面的代码，它“工作”...so很远。所谓“工作”，我的意思是Flux<DemoPOJO>由service.getAll()返回，而"hasElements().subscribe(this::foo)“生成的foo()生成的输出正确地反映了Flux<DemoPOJO>是否有任何元素。
所需的结束状态是返回一个ServerResponse对象，包装Flux<DemoPOJO>，这反映返回的Flux<DemoPOJO>是空的还是"hasElements“的。
我的问题是Mono.subscribe()返回一个reactor.co
浏览 1提问于2019-07-22得票数 0
回答已采纳
根据我到目前为止所读到的内容，可以使用ConnectableFlux将Flux多播到多个订阅者，方法类似于以下内容：
Flux<Integer> integerFlux = Flux.range(1, 50)
.log();
ConnectableFlux<Integer> integerConnectableFlux = integerFlux.publish();
integerConnectableFlux.map(i -> i*2)
.subscribe(System.out::print
浏览 14提问于2021-10-21得票数 1
2回答
嗨，我有Flux，在每个元素的迭代过程中，它创建了新的mono。在流动之外，我还有其他的单极。并想要做以下工作:当通量(与相应的内部单体的末端)，然后做第二单。challegeing是mono内部从webclient请求中创建的流量。作为起点，请看“加载”方法。基本没有webclient，它可以工作，但如果与webclient在地图工作之后的工作。使用弹簧引导2
public WebClient.ResponseSpec sendGetRequest(String path, Object... pathVariables){
try {
LOGGER.info(&
浏览 0提问于2019-05-03得票数 0
回答已采纳
我使用的是反应堆核心。
考虑下面的代码片段：
Flux<String> flux = Flux.<String>create(sink -> sink.next("test"))
.replay(1)
.refCount();
flux.subscribe(System.out::println);
flux.next().subscribe(System.out::println); // The exception is thrown here!
预期输出：
test
test
实际输出：
test
Exception in
浏览 2提问于2018-02-20得票数 2
回答已采纳
在博客文章中，作者建议使用subscribeOn调用将同步阻塞调用包装在Mono中，如文章中的以下代码片段所示：
final Flux<String> betterFetchUrls(List<String> urls) {
return Flux.fromIterable(urls)
.flatMap(url ->
//wrap the blocking call in a Mono
Mono.fromCallable(() -> blockingWebClient.get(url))
浏览 13提问于2020-12-10得票数 0
1回答
在上下文中失败。
、、
我想在我的通量管道中使用Context来绕过过滤。
我现在拥有的是：
public Flux<Bar> realtime(Flux<OHLCIntf> ohlcIntfFlux) {
return Flux.zip(
ohlcIntfFlux,
ohlcIntfFlux.skip(1),
Mono.subscriberContext().map(c -> c.getOrDefault("isRealtime", false))
浏览 1提问于2019-04-07得票数 1
回答已采纳
我有一个聚合工具类，其中我必须连接多个cassandra表数据。我的生产代码如下所示，但不完全相同。
@Autowired FollowersRepository followersRepository;
@Autowired TopicRepository topicRepository;
@GetMapping("/info")
public Flux<FullDetails> getData(){
return Flux.create(emitter ->{
浏览 0提问于2018-08-16得票数 1
回答已采纳
1回答
单螺纹单焊剂
、、、、
我有来自db的Flux<Foo> (例如，5个元素)。我需要从每个Foo获取一些信息，将其全部设置为Mono<MyRequest>，发送到另一个rest资源，获取一个Mono<MyResponse>并在每个Foo中使用其中的所有信息。
我在Flux.flatMap()中使用了大量的Mono.zipWith()和Mono.zipWhen()、，但是创建MyRequest和发送到资源的情况由5个线程执行了5次。
Flux<Foo> flux = dao.getAll();
Flux<Foo> fluxAfterProcessing = f
浏览 0提问于2018-02-05得票数 3
回答已采纳
1回答
我有一个通量和Mono，如下所示：
Mono<MyRequest> req = request.bodyToMono(MyRequest.class);
Mono<List<String>> mono1 = req.map(r -> r.getList());;
Flux<Long> flux1 = req.map(r -> r.getVals()) // getVals() return list of Long
.flatMapMany(Flux::fromIterable);
现在，对于flux1中的每个数字，我
浏览 0提问于2018-08-17得票数 1
回答已采纳
刚刚开始理解使用反应堆进行反应性编程，我在这里遇到了一个教程中的代码片段--
class ChatSocketHandler(val mapper: ObjectMapper) : WebSocketHandler {
val sink = Sinks.replay<Message>(100);
val outputMessages: Flux<Message> = sink.asFlux();
override fun handle(session: WebSocketSession): Mono<Void> {
浏览 6提问于2021-05-20得票数 2
回答已采纳
1回答
我有一个长期运行的Flux，并希望登录1在N元素，以监测进展。下面的代码以N毫秒为单位记录一段代码。
Flux
.fromStream(
IntStream
.range(1, 101)
.mapToObj(Integer::valueOf)
)
.sample(Duration.ofMillis(2))
.subscribe(e -> log.debug(e.toString()));
听起来，可以通过为1元素生成Mono.Just("")和为其余元素生成Mono.empty
浏览 4提问于2021-04-18得票数 0
回答已采纳
我不知道如何使用当时的项目反应堆操作员。我不太理解这个操作符的javadoc。谁能向我解释这件事？有了一些代码示例就很好了。我自己写了一个例子如下：
public class Test11 {
public static void main(String[] args) {
Flux flux = Flux.just(1, 2, 3);
Mono<Void> mono1 = flux.then();
mono1.subscribe(null, null, () -> System.out.println("yes
浏览 0提问于2021-05-10得票数 0
回答已采纳
1回答
我有一个Flux流。对于每个被处理的元素，我希望触发一个异步/非阻塞的操作。例如，从数据库更新返回Mono的方法。我希望这个操作在doOnNext块上完成。我不想影响在那里实现的Flux、处理和背压。 假设要调用的Mono方法是 Mono<Integer> dbUpdate(); 我的Flux应该是这样的吗？ public Flux<Data> processData(PollRequest request)
{
return searchService.search(request)
浏览 33提问于2021-11-09得票数 0
回答已采纳
我希望在Mono处于活动状态时，每秒钟执行一次操作。做这件事最好的方法是什么？
这是一个可行的选择，但它似乎是一个解决办法：
// Some long life async action
Mono<String> asyncAction = Mono.delay(Duration.ofSeconds(60)).map(d -> "Hello");
Mono<String> periodicAction = Flux.interval(Duration.ofSeconds(1))
.doOnNext(d -> {
浏览 6提问于2021-12-12得票数 0
回答已采纳
2回答
我有一个为kubernetes调用和检测服务可用的api，或者not.In该api，首先调用一个接口来获取其他服务的主机，接口返回一个Flux，如果结果是空api返回SERVICE_UNAVAILABLE其他返回ok.My当前代码如下：
@GetMapping(value = "/gateway/readiness")
public Mono<Long> readiness(ServerHttpResponse response) {
Flux<Map<String, List<String>>> hosts = host
浏览 9提问于2020-12-03得票数 0
回答已采纳
我的端点是这样的:-
public Mono<Void> postData( Flux<SomeModelClass> dataFlux ){
...
repository.saveAll(dataFlux);
...
}
由于这并不是将“dataFlux”保存到我的数据库中，所以我不得不将返回类型和方法定义更改为：
public Flux<SomeModelClass> postData( Flux<SomeModelClass> dataFlux ){
...
return repository.saveAll(dataFlux);
...
浏览 1提问于2022-10-31得票数 0
回答已采纳
我是java的新手。我试图为我的API实现更新方法。我想更新我的数据的一些字段。以下是我的数据类：
public class Task implements Serializable {
@Id
private String id;
private String task;
private Boolean isCompleted;
private String userId;
}
我的服务接口：
public interface TaskService {
void create(Task task);
Mono<Task&g
浏览 9提问于2021-12-09得票数 0
回答已采纳
我想“分享”一个Mono，就像我对Flux做的一样。
通量共享()示例与Kotlin：
fun `test flux share`() {
val countDownLatch = CountDownLatch(2)
val originalFlux = Flux.interval(Duration.ofMillis(200))
.map { "$it = ${Instant.now()}" }
.take(7)
.share()
浏览 1提问于2019-07-07得票数 2
回答已采纳
我正在爬行一个API并构建以下数据结构：
public class Organization {
String id;
String name;
Flux<Space> spaces;
}
public class Space {
String id;
String name;
Flux<Application> applications;
}
public class Application {
String id;
String name;
Flux<Process> proce
浏览 0提问于2019-08-04得票数 0
回答已采纳
1回答
我有以下场景:我有一个ProductFamily，它有一组产品。
我从数据库加载ProduktFamily，然后加载它的产品并将它们插入到ProduktFamily.products中。
在Spring中很容易(因为有了JPA)，但是对于Webflux，我正在挣扎。我试过了，但不起作用。产品系列产品是空的。
服务：
public Flux<ProduktfamilyRS> getAllProduktFamilies() {
return produktfamilyDatabaseFacade.findAll()
.map(produktfamilyEntit
浏览 3提问于2021-08-02得票数 0
使用Reactor，我有一个Mono和一个Flux，Mono在一个通道上做一些轮询，Flux在这个通道上发布。在一个测试中，我希望Flux仅在Mono有效轮询时才开始发布数据。由于Mono上的轮询不会在订阅后立即启动，因此在开始发布之前，我一直在使用固定的delaySubscription：
Mono<...> polling;
Flux<...> dataPublisher;
polling
.zipWith(dataPublisher.collectList().delaySubscription(Duration.ofSeconds(1)))
浏览 1提问于2021-03-13得票数 0
1回答
我在Flux后面附加了一个flatMap，但是如果我添加额外的flatMaps，只会返回最后一个。
// Here is an example of the Mono function
private Mono<MyType> appendFirstMono(Group group) {
return Mono.just(group)
.map(MyType::new)
.flatMap(g -> functionZ(group)
.map(g::setField));
}
//This works as expecte
浏览 1提问于2018-12-28得票数 2
我有一个webflux应用程序。在一个地方，我进行了3次API调用。每次调用都依赖于上一次调用的结果。 Mono<List<Book>> books = bookService.find(params); //1st API call
Mono<List<User>> users = books.flatMapMany(Flux::fromIterable)
.map(User::getId)
.collectList()
.map(allUserIds -> userService.findUsers(allUs
浏览 88提问于2021-07-26得票数 0
我有一个方法，它遍历购物车中的项目，并使用placeOrder为相同的项目下订单。对购物车中的所有商品调用placeOrder后，我希望合并并发送一个Mono对象，总结哪些订单完成了哪些订单没有完成 这段代码可以工作，但没有使用placeOrder的并行执行。 List<Mono<OrderResponse>> orderResponse = new ArrayList<Mono<OrderResponse>>();
OrderCombinedResponse combinedResponse = new OrderCombin
浏览 58提问于2019-06-13得票数 0
getUserDetails方法返回JsonNode类型的Mono。，但实际上我想返回一个Mono<User.java>或Flux<User.java>。请帮助修改getBulkUserInfo或getUserDetails以获得Mono<User.java>或Flux<User.java>
public Mono<JsonNode> getUser(BigInteger Id){
return this.client.get()
.uri("/URL/{Id}",Id)
浏览 3提问于2022-01-18得票数 0
回答已采纳
1回答
我有一个类，它从Flux源填充Map。我希望能够动态地进行updateTheData()，但我希望getTheData()能够返回一个解析为当前数据或挂起数据的Mono。
基本上，如果theData != null和no正在进行中，则返回Mono.just( theData )，否则返回最终将发出theData的mono。
编辑:这是到目前为止我所能做的最好的。
class SomeClass {
private Mono<Map<String, Data>> theData;
private final SomeFluxServ
浏览 2提问于2019-10-16得票数 2
回答已采纳
我正在努力为我的工作实现反应堆核心。我被困在重试，我们需要在错误的情况下执行。下面是我添加错误之前的示例代码
FluxSink<String> mainSink;
// Create the fulx and get handle to Sink
Flux<String> mainFlux = Flux.create(sink -> {
mainSink = sink;
}, FluxSink.OverflowStrategy.BUFFER);
// Convert to Hot Flux
ConnectableFlux<String> hotF
浏览 0提问于2018-08-22得票数 3
回答已采纳
我有以下方法：
public void countLetters() {
Flux.just("alpha", "bravo", "charlie")
.map(String::toUpperCase)
.flatMap(s -> Flux.fromArray(s.split("")))
.groupBy(String::toString)
.sort(Comparator.comparing(GroupedFlux::k
浏览 5提问于2021-11-05得票数 0
回答已采纳
当其他Mono为空时，我有一个要修改的Mono。
第一Mono
deviceRepository.getDevice(deviceId, userId): Mono<Device>
二次Mono
registrationRepository.canRegister(deviceId) : Mono<DeviceRegistrationInfo>
我想要实现的是，当Mono<Device>为给定的deviceId返回空的mono时，返回修改后的deviceId：
device.copy(status = DeviceStatus.REGISTRATION_FA
浏览 5提问于2018-11-25得票数 0
回答已采纳
reactor和spring webflux的新手，发现范式转换相当困难。 我有一个Mono<OAuth2User>，它有一个Map<String, Object> getAttributes(); 我想检查一下地图上的内容。 我已经把它分解为只是试图在Mono中处理一张地图。 Mono<Map> mm = Mono.just(Map.of("a",1,"b",2,"c",3))
mm.log().map(m -> m.entrySet()).flatMapMany(Flux::fromIterable
浏览 89提问于2019-04-30得票数 1
回答已采纳
我试图用Mono中的值填充Flux中的对象。当我尝试这样做的时候，它只是忽略了我的"set“操作。我认为这是因为Flux是并行工作的，而Mono不是。我该如何解决这个问题？ Flux.fromIterable(proxyParserService.getProxyList())
.parallel()
.runOn(Schedulers.parallel())
.filter(proxy -> proxy.getCorrupted() == null || !proxy.getCorrupted())
浏览 32提问于2020-01-09得票数 0
1回答
我是反应性编程方面的新手。我看到可以压缩两个monos以生成一个结果：
Mono<Info> info = Mono.just(id).map(this::getInfo).subscribeOn(Schedulers.parallel());
Mono<List<Detail>> detail= Mono.just(petitionRequest).map(this.service::getDetails)
.subscribeOn(Schedulers.parallel());
Flux<Generate
浏览 0提问于2019-07-22得票数 2
回答已采纳
新的反应堆，试图理解Mono.flatMapMany和Mono.flatMapIterable经常处理包含集合的对象的Mono。
在以下方面有何不同：
Mono.just(List.of("one", "two", "three")).log().flatMapMany(Flux::fromIterable).subscribe(System.out::println);
Mono.just(List.of("one", "two", "three")).log().
浏览 1提问于2019-05-03得票数 5
回答已采纳
我是WebClient和响应式编程的新手。我想从一个请求中获得响应体。如果出现错误，必须记录http-code、头和正文，但仍应返回正文。 经过大量的挖掘和谷歌搜索，我找到了两个解决方案。但在我看来，这两种方法都很复杂。有没有更简单的解决方案？ 继续使用Mono，我找到了这个解决方案： public Mono<String> log(ProtocolLine protocolLine) {
return webClient.post()
.uri("/log")
.body(BodyInserters.fro
浏览 68提问于2019-06-27得票数 2
回答已采纳
所以现在，我会给出一个回复
@GetMapping("/integers")
@ResponseStatus(code = HttpStatus.OK)
public Mono<Map<String, Flux<Integer>>> getIntegers() {
Mono<Map<String, Flux<Integer>>> integers =
Mono.just(Map.of("Integers", intege
浏览 0提问于2018-07-11得票数 2
回答已采纳
我在spring web flux中使用web reactive。我已经为POST请求实现了一个处理程序函数。我要服务器立即返回。因此，我实现了如下的处理程序-：
public class Sample implements HandlerFunction<ServerResponse>{
public Mono<ServerResponse> handle(ServerRequest request) {
Mono bodyMono = request.bodyToMono(String.class);
bodyMono.map(str -> {
Sy
浏览 0提问于2019-09-03得票数 0
1回答
我正在使用WebFlux，我想用AOP登录，如下所示： Class LogAop {
@AfterReturning("execution ...")
public void configSetted() {
// Getting username by token
Mono<Sting> username = ReactiveSecurityContextHolder.getContext()
.map { ... };
username.subscribe ({it -> loggerin.in
浏览 101提问于2021-11-08得票数 0
3回答
我正在使用Spring 5 Webflux和Kotlin编写一个简单的应用程序。我正在尝试通过以下方式实现PUT端点：
PUT("/confs/{id}", {
val id = it.pathVariable("id")
ServerResponse.ok().body(service.save(it.bodyToMono(Item::class.java)), Item::class.java)
})
保存的诀窍是，我尝试从项目中读取一个城市名称，解析地理坐标，在原始项目中覆盖它们，然后使用Spring Data Mongo Reactive
浏览 17提问于2018-01-01得票数 10
回答已采纳
在文档中，您应该将阻塞代码包装到一个Mono：中
但它并没有写出如何真正做到这一点。
我有以下代码：
@PostMapping(path = "some-path", consumes = MediaType.APPLICATION_STREAM_JSON_VALUE)
public Mono<Void> doeSomething(@Valid @RequestBody Flux<Something> something) {
something.subscribe(something -> {
// some blocki
浏览 2提问于2018-08-29得票数 2
回答已采纳
我有无限的热流数据。我将要在流中的每个元素上执行一个操作，每个元素都返回一个Mono，它将在一段有限的时间后(以某种方式)完成。 这些操作可能会抛出错误。如果是这样，我希望重新订阅热点而不会遗漏任何东西，重试抛出错误时正在处理中的元素(即任何未成功完成的元素)。 我在这里做什么？我可以容忍对相同元素的重复操作，但不会完全丢失流中的元素。 我尝试过使用ReplayProcessor来处理这个问题，但是我看不到一种方法可以让它工作，而不是重复许多可能成功的操作(使用非常保守的超时)，或者由于新元素覆盖缓冲区中的旧元素而丢失元素(如下所示)。 测试用例： @Test
public
浏览 11提问于2019-09-05得票数 3
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
