将Spring Integration executorChannel与Spring Cloud函数结合使用

Question

我使用spring-cloud函数来处理来自Flux的kafka的数据。默认情况下，它在使用者线程(消息被消费的地方(中处理数据。我将实现用于并行数据处理和节流的线程池，Spring Cloud Integration中有一个很棒的实现，叫做executorChannel(https://docs.spring.io/spring-integration/api/org/springframework/integration/channel/ExecutorChannel.html)

功能实现示例：

public static class FN1 implements Function<Flux<String>, Flux<String>> {
public Flux<String> apply(Flux<String> data) {
return data
.map(f ->  doSomething() )      
}
}

因此，我找不到通过executorChannel连接以这种方式实现的函数的简单方法。

M.b.有一种方法可以定义inputChannel类型吗？

UPD：阅读奥列格回答下的评论。它们非常有用。

Answer 1

你的意思是这样的吗？

@SpringBootApplication
public class SampleFunctoinAppApplication  {
public static void main(String[] args) throws Exception {
ApplicationContext context = SpringApplication.run(SampleFunctoinAppApplication.class, args);
SubscribableChannel output = context.getBean("output", SubscribableChannel.class);
output.subscribe(System.out::println);
MessageChannel channel = context.getBean("executorChannel", MessageChannel.class);
channel.send(new GenericMessage<String>("hello"));
}
@Bean
public IntegrationFlow flow() {
return IntegrationFlows
.from("executorChannel")
.transform(echo())
.channel("output")
.get();
}
@Bean
public ExecutorChannel executorChannel() {
return new ExecutorChannel(Executors.newCachedThreadPool());
}
public Function<String, String> echo() {
return v -> v;
}
}

您所说的"定义inputChannel类型"是什么意思？