嗨,我有助焊剂,在每个元素上都创建新的单声道。我还在磁通范围外还有其他单声道。并想执行以下操作:当磁通(带有相应的内部单声道的末端(时,请执行第二个单声道。Challegeing Prart是Mono内部的磁通量是根据WebClient请求创建的。作为起点,请查看"负载"方法。 基本上没有网络电量,它可以正常工作,但是在此之后使用WebClient内部地图工作。 使用Spring-boot 2
public WebClient.ResponseSpec sendGetRequest(String path, Object... pathVariables){
try {
LOGGER.info("content type {}, url {}, path {}", contentType, url, path);
WebClient.ResponseSpec responseSpec = sendRequest(HttpMethod.GET, contentType, authorizationToken, url, path, pathVariables);
return responseSpec;
}catch (Exception e){
throw new WebClientProcessingException("Exception when trying to process", e);
}
}
public Mono<PersonPayload> loadPerson(String path){
try {
LOGGER.info("path {}", path);
Mono<QuestionDetailsPayload> person = sendGetRequest(path).bodyToMono(PersonPayload.class);
return person;
}catch (Exception e){
throw new WebClientProcessingException("Exception when trying to process",e);
}
}
public Mono<PersonDomain> getPerson(String path) {
Assert.notNull(path, "path can't be null");
try{
LOGGER.info("path {}" ,path);
Mono<PersonPayload> personPayload = loadPerson(path);
return personPayload.map(this::toPersonDomain);
}catch (Exception e){
throw new PersonNotFoundException("Exception when trying to get person info" , e);
}
}
public PersponDomain toPersonDomain(PersonPayload personPayload){
return modelMapper.map(personPayload, PersonDomain.class);
}
public void load(){
List<String> outStr = Arrays.asList("out1", "out2","out3");
Flux flux = Flux.fromIterable(outStr);
Flux<Mono<PersonDomain>> results = flux.map(string ->{
System.out.println(string);
Mono<PersonDomain> personMono = getPerson("inside");
Mono<String> result = personMono.map(h ->{
System.out.println(personMono.getName());
return personMono.getName() + "_test";
});
return result;
});
Mono<String> second = Mono.just("second");
results.then(second);
results.subscribe(stringMono -> {
stringMono.subscribe();
});
second.subscribe( s->{
System.out.println(s);
});
}
gradle依赖性:
implementation 'org.springframework.boot:spring-boot-starter-webflux'
implementation 'org.springframework.boot:spring-boot-starter-validation'
implementation 'org.postgresql:postgresql'
implementation 'org.springframework.boot:spring-boot-starter-jooq'
implementation 'org.jooq:jooq-codegen'
implementation 'org.modelmapper:modelmapper:2.3.0'
implementation 'org.modelmapper:modelmapper:2.3.0'
compileOnly 'org.projectlombok:lombok'
annotationProcessor 'org.projectlombok:lombok'
implementation 'com.google.code.gson:gson:2.8.5'
testImplementation 'org.springframework.boot:spring-boot-starter-test'
testImplementation 'io.projectreactor:reactor-test'
testImplementation 'org.powermock:powermock-module-junit4:2.0.0'
testImplementation 'org.powermock:powermock-api-mockito2:2.0.0'}
Flux#map
是同步操作,不订阅返回的对象。
您应该使用Flux#flatMap
/Flux#concatMap
/Flux#flatMapSequential
/Flux#switchMap
。这些操作员将订阅返回的Publisher
。
解决我的情况的解决方案使用:
subscribe(Consumer<? super T> consumer,
Consumer<? super Throwable> errorConsumer,
Runnable completeConsumer);
而不是这样。处理值和错误,但在序列成功完成后也执行一些代码。