我有一個聚合函式，可以聚合發布到output-channel. 我已經訂閱了如下函式生成的通量：
@Component
public class EventsAggregator {
@Autowired
private Sinks.Many<Message<?>> eventsPublisher; // Used to publish events from different places in the code
private final Function<Flux<Message<?>>, Flux<Message<?>>> aggregatorFunction;
@PostConstruct
public void aggregate() {
Flux<Message<?>> output = aggregatorFunction.apply(eventsPublisher.asFlux());
output.subscribe(this::aggregate);
}
public void aggregate(Message<?> aggregatedEventsMessage) {
if (...) {
//some code
} else {
throw new RuntimeException();
}
}
}
如果RuntimeException拋出 ，則聚合函式不起作用，我收到此訊息The [bean 'outputChannel'; defined in: 'class path resource [org/springframework/cloud/fn/aggregator/AggregatorFunctionConfiguration.class]'; from source: 'org.springframework.cloud.fn.aggregator.AggregatorFunctionConfiguration.outputChannel()'] doesn't have subscribers to accept messages at org.springframework.util.Assert.state(Assert.java:97)
有什么辦法可以安全的訂閱聚合函式產生的流量嗎？
uj5u.com熱心網友回復：
這是正確的。這就是 Reactive Streams 的作業原理：如果拋出例外，訂閱者將被取消，并且不能再向該訂閱者發送新資料。
考慮不要將該例外拋出到流中。
在檔案中查看更多資訊：https ://docs.spring.io/spring-cloud-stream/docs/4.0.0-SNAPSHOT/reference/html/spring-cloud-stream.html#spring-cloud-stream-overview-error -處理
轉載請註明出處，本文鏈接：https://www.uj5u.com/caozuo/534311.html