如何通过Direct Channel获取Spring Integration流程的响应？

我通过Java类触发基于XML配置的HTTP REST API流程，代码如下：

@Autowired
@Qualifier("request.channel")
MessageChannel httpOutboundGateway;


Message<?> message = MessageBuilder.withPayload("")
                    .setHeader("bstUrl", "http://test-core-api-wcnp.stg.com/api/v1/workflow?userId=\"1234\"&userType=\"abc\"")
                    .setHeader("userId", "1234")
                    .setHeader("userType", "abc")
                    .build();

httpOutboundGateway.send(message);
            
/* MessagingTemplate template = new MessagingTemplate();
Message reply = template.sendAndReceive(httpOutboundGateway, message); */

XML配置如下：

<int:channel id="request.channel" />
<int:channel id="response.channel" />

<int:gateway id="requestGateway"
             default-request-channel="request.channel">
</int:gateway>

<int:chain id="workload-chain" input-channel="request.channel" output-channel="response.channel">
    <int-http:outbound-gateway id="workflowGateway"
                               url-expression="headers.bstUrl"
                               http-method="GET"
                               expected-response-type="java.lang.String"
                               charset="UTF-8"
                               request-factory="httpComponentClientRequestFactory"
                               mapped-request-headers="*" >
  </int-http:outbound-gateway>
        <int:transformer ref="kafkaTransformer" method="fetchWorkflowCoreServiceResponse" />
</int:chain>

我在fetchWorkflowCoreServiceResponse方法中把String响应转换为所需DTO类，但后续逻辑需要在发送消息后获取该响应。尝试调用response.channel的receive方法，但该方法需要Pollable Channel，而我用的是Direct Channel；使用MessagingTemplate的sendAndReceive方法时，报错“dispatcher has no subscribers for response channel”。请问如何从response.channel按需获取响应？

方案1：利用Spring Integration Gateway实现请求-响应模式 #

这是最简洁的同步获取响应方式，你已定义requestGateway，只需补充接口和配置：

1. 定义Gateway接口：

public interface WorkflowGateway {
    YourDtoClass sendRequest(Message<?> message);
}

2. 修改XML中的requestGateway配置，指定接口和回复通道：

<int:gateway id="requestGateway"
             service-interface="com.your.package.WorkflowGateway"
             default-request-channel="request.channel"
             default-reply-channel="response.channel">
</int:gateway>

3. 在业务代码中注入接口并直接调用：

@Autowired
private WorkflowGateway requestGateway;

// 构建message代码不变
YourDtoClass responseDto = requestGateway.sendRequest(message);

Spring Integration会自动处理请求-响应的关联逻辑，无需手动管理通道订阅。

方案2：将response.channel改为Pollable Channel #

如果必须用receive方法获取响应，把response.channel配置为队列通道（Pollable Channel的实现）：

1. 修改XML通道配置：

<int:channel id="response.channel">
    <int:queue />
</int:channel>

2. 在代码中注入PollableChannel并调用receive：

@Autowired
@Qualifier("response.channel")
private PollableChannel responseChannel;

// 发送消息后获取响应
httpOutboundGateway.send(message);
Message<?> responseMessage = responseChannel.receive(5000); // 超时时间，单位毫秒
if (responseMessage != null) {
    YourDtoClass responseDto = (YourDtoClass) responseMessage.getPayload();
    // 后续业务逻辑处理
}

注意：需确保消息发送和接收的线程同步，避免响应被其他消费者取走。

方案3：修复MessagingTemplate的sendAndReceive用法 #

之前报错是因为sendAndReceive要求回复通道有订阅者，正确用法需指定默认回复通道：

@Autowired
private MessagingTemplate messagingTemplate;

@Autowired
@Qualifier("request.channel")
private MessageChannel requestChannel;

// 设置默认回复通道
messagingTemplate.setDefaultReplyChannelName("response.channel");
Message<?> reply = messagingTemplate.sendAndReceive(requestChannel, message);
if (reply != null) {
    YourDtoClass responseDto = (YourDtoClass) reply.getPayload();
}

此方式不如方案1简洁，推荐优先使用Gateway接口实现。

内容的提问来源于stack exchange，提问作者KeepItSimple