在 Spring Cloud Stream 的单个处理器中执行记录聚合并启动 Spring Cloud 任务

Question

我正在尝试执行以下操作

1. 聚合消息
2. 启动 Spring Cloud 任务 但无法将聚合消息传递给启动 Task 的方法。下面是一段代码

    @Autowired
    private TaskProcessorProperties processorProperties;

    @Autowired
    Processor processor;

    @Autowired
    private AppConfiguration  appConfiguration ;


    @Transformer(inputChannel = MyProcessor.intermidiate, outputChannel = Processor.OUTPUT)
    public Object setupRequest(String message) {
        Map<String, String> properties = new HashMap<>();

        if (StringUtils.hasText(this.processorProperties.getDataSourceUrl())) {
            properties.put("spring_datasource_url", this.processorProperties.getDataSourceUrl());
        }
        if (StringUtils.hasText(this.processorProperties.getDataSourceDriverClassName())) {
            properties.put("spring_datasource_driverClassName", this.processorProperties
                    .getDataSourceDriverClassName());
        }
        if (StringUtils.hasText(this.processorProperties.getDataSourceUserName())) {
            properties.put("spring_datasource_username", this.processorProperties
                    .getDataSourceUserName());
        }
        if (StringUtils.hasText(this.processorProperties.getDataSourcePassword())) {
            properties.put("spring_datasource_password", this.processorProperties
                    .getDataSourcePassword());
        }
        properties.put("payload", message);

        TaskLaunchRequest request = new TaskLaunchRequest(
                this.processorProperties.getUri(), null, properties, null,
                this.processorProperties.getApplicationName());
        System.out.println("inside task launcher  **************************");
        System.out.println(request.toString() +"**************************");
        return new GenericMessage<>(request);
    }


    @ServiceActivator(inputChannel = Processor.INPUT,outputChannel = MyProcessor.intermidiate)
    @Bean
    public MessageHandler aggregator() {
        AggregatingMessageHandler aggregatingMessageHandler =
                new AggregatingMessageHandler(new DefaultAggregatingMessageGroupProcessor(),
                        new SimpleMessageStore(10));
        AggregatorFactoryBean aggregatorFactoryBean = new AggregatorFactoryBean();
        //aggregatorFactoryBean.setMessageStore();
        //aggregatingMessageHandler.setOutputChannel(processor.output());
        //aggregatorFactoryBean.setDiscardChannel(processor.output());
        aggregatingMessageHandler.setSendPartialResultOnExpiry(true);
        aggregatingMessageHandler.setSendTimeout(1000L);
        aggregatingMessageHandler.setCorrelationStrategy(new ExpressionEvaluatingCorrelationStrategy("'FOO'"));
        aggregatingMessageHandler.setReleaseStrategy(new MessageCountReleaseStrategy(3)); //ExpressionEvaluatingReleaseStrategy("size() == 5")
        aggregatingMessageHandler.setExpireGroupsUponCompletion(true);
        aggregatingMessageHandler.setGroupTimeoutExpression(new ValueExpression<>(3000L)); //size() ge 2 ? 5000 : -1
        aggregatingMessageHandler.setExpireGroupsUponTimeout(true);
        return aggregatingMessageHandler;
    }

为了在聚合器和任务启动器方法 (setupRequest(String message)) 之间传递消息，我使用了一个定义如下的通道 MyProcessor.intermidiate

import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.stereotype.Indexed;


public interface MyProcessor {
    String intermidiate = "intermidiate";

    @Output("intermidiate")
    MessageChannel intermidiate();

}

使用的Applicaion.properties如下

aggregator.message-store-type=persistentMessageStore
spring.cloud.stream.bindings.input.destination=output
spring.cloud.stream.bindings.output.destination=input

它不起作用，使用上述方法。

在此类中，如果我将通道名称从我定义的通道 MyProcessor.intermediate 更改为 Processor.input 或 Processor.output，那么任何一项都可以正常工作（基于更改为 Processor.* 的通道名称）

我想先聚合消息，而不是想在处理器中启动聚合消息的任务，这不会发生

Answer 1

看这里：

public Object setupRequest(String message) {

因此，您期望一些 string 作为请求负载。

您的AggregatorFactoryBean 使用DefaultAggregatingMessageGroupProcessor，它正是这样做的：

    List<Object> payloads = new ArrayList<Object>(messages.size());
    for (Message<?> message : messages) {
        payloads.add(message.getPayload());
    }
    return payloads;

所以，肯定不是String。

奇怪的是，您没有显示您的配置发生了什么异常，但我假设您需要更改 setupRequest() 签名以期望有效负载的 List 或者您需要提供一些自定义 MessageGroupProcessor 来构建String 来自您聚合的消息组。