java - 带有 RestController 的 Spring IntegrationFlow

问题描述

我想创建一个执行以下步骤的应用程序：

通过 RestController 接收请求
将接收到的消息发送到队列（AMQP - MessageChannel）（correlationId？）
等待另一个队列中的回复（AMQP - MessageChannel）（correlationId？）
在与步骤 1 中的请求相同的线程上返回响应。

我考虑过为此使用 IntegrationFlow，但我无法调整这些步骤。

此外，您知道实现此流程的最佳方法是什么吗？

我尝试使用以下代码实现：

package poc.integration.http;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.http.HttpMethod;
import org.springframework.integration.amqp.channel.PollableAmqpChannel;
import org.springframework.integration.amqp.dsl.Amqp;
import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint;
import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.http.dsl.Http;
import org.springframework.integration.http.inbound.HttpRequestHandlingMessagingGateway;
import org.springframework.integration.http.inbound.RequestMapping;
import org.springframework.integration.http.outbound.HttpRequestExecutingMessageHandler;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.messaging.MessageChannel;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;

@Configuration
@EnableIntegration
public class IntegrationFlowConfig {

    final Logger logger = LoggerFactory.getLogger(IntegrationFlowConfig.class);

    @Bean
    public HttpRequestHandlingMessagingGateway inbound() {
        HttpRequestHandlingMessagingGateway gateway = new HttpRequestHandlingMessagingGateway(true);
        gateway.setRequestMapping(mapping());
        gateway.setRequestPayloadTypeClass(String.class);
        gateway.setRequestChannelName("httpRequest");
        return gateway;
    }

    @Bean
    public RequestMapping mapping() {
        RequestMapping requestMapping = new RequestMapping();
        requestMapping.setPathPatterns("/foo");
        requestMapping.setMethods(HttpMethod.GET);
        return requestMapping;
    }

    @ServiceActivator(inputChannel = "httpResponse")
    @Bean
    public HttpRequestExecutingMessageHandler outbound() {
        HttpRequestExecutingMessageHandler handler =
            new HttpRequestExecutingMessageHandler("http://10.141.201.206:80/foo");
        handler.setHttpMethod(HttpMethod.GET);
        handler.setExpectedResponseType(String.class);
        return handler;
    }

    @ServiceActivator(inputChannel="httpRequest", outputChannel="httpResponse")
    public Object processarMensagem(Object mensagem) {
        return mensagem + " - done";
    }

    @Bean(name = PollerMetadata.DEFAULT_POLLER)
    public PollerMetadata pollerAmqp(ThreadPoolTaskScheduler taskScheduler) {
        final PollerMetadata poller = new PollerMetadata();
        poller.setTaskExecutor(taskScheduler);
        poller.setReceiveTimeout(-1);
        return poller;
    }

    @Bean
    public MessageChannel httpRequest(AmqpTemplate amqpTemplate) {
        PollableAmqpChannel channel = new PollableAmqpChannel("httpRequest", amqpTemplate,
                DefaultAmqpHeaderMapper.outboundMapper(), DefaultAmqpHeaderMapper.inboundMapper());
        channel.setExtractPayload(true);
        return channel;
    }

    @Bean
    public MessageChannel httpResponse(AmqpTemplate amqpTemplate) {
        PollableAmqpChannel channel = new PollableAmqpChannel("httpResponse", amqpTemplate,
                DefaultAmqpHeaderMapper.outboundMapper(), DefaultAmqpHeaderMapper.inboundMapper());
        channel.setExtractPayload(true);
        return channel;
    }

}

但我收到消息：超时内未收到回复

标签： javaspringspring-mvcspring-integration

你只需要一个IntegrationFlow和一个Http.inboundControllerAdapter()作为起点。@RestController它完全取代了提到的 RestController，但让您避免额外的工作IntegrationFlow。

流程的下一步应该是Amqp.outboundGateway()通过 AMQP 发送和接收。这会为您处理相关性。

https://docs.spring.io/spring-integration/docs/5.3.0.RELEASE/reference/html/amqp.html#configuring-with-the-java-dsl-4

java - 带有 RestController 的 Spring IntegrationFlow

问题描述

解决方案

推荐阅读