Spring Integration应用如何迭代调用outbound-gateway实现分页拉取API数据？

实现方案 #

完全可以通过路由器组件搭建循环分页调用机制，整套流程的改造步骤如下：

1. 改造HTTP出站网关的分页参数配置 #

首先将原网关URL中写死的page=1改为从消息头动态取值，新增分页参数的URI变量绑定：

<int-http:outbound-gateway
            url="${xxx.smartbrain.produto}?startDate={dtInitPosicao}&amp;endDate={dtEnndPosicao}&amp;page={currentPage}&amp;recordPerPage=15&amp;cblc={cblc}&amp;cpfCnpj={cpf}"
            http-method="GET"
            expected-response-type="br.com.bradesco.ciar.xxx.xxx.xxx.integration.response.ResponseTesouroXXXXX"
            request-channel="publishSubRequest"
            reply-channel="gatewayResponseProcessChannel"
            rest-template="restTemplate"
            mapped-request-headers="Authorization, cpf, cblc, dtInitPosicao, currentPage">
        <int-http:uri-variable name="cpf" expression="headers.cpf"/>
        <int-http:uri-variable name="cblc" expression="headers.cblc"/>
        <int-http:uri-variable name="dtInitPosicao" expression="headers.dtinitposicao"/>
        <int-http:uri-variable name="currentPage" expression="headers.currentPage"/>
</int-http:outbound-gateway>

注意：首次发起请求时需要给初始消息新增currentPage=1的请求头

2. 新增分页响应预处理节点 #

新增服务激活节点，负责提取响应中的总页数、累加当前页码、聚合已拉取的分页数据：

<int:service-activator 
    input-channel="gatewayResponseProcessChannel" 
    output-channel="pageCheckRouterChannel" 
    ref="pageProcessHandler" 
    method="handle"/>

对应Java处理类实现：

import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import java.util.ArrayList;
import java.util.List;

public class PageProcessHandler {
    public Message<ResponseTesouroXXXXX> handle(Message<ResponseTesouroXXXXX> message) {
        ResponseTesouroXXXXX currentPage = message.getPayload();
        Integer currentPageNum = message.getHeaders().get("currentPage", Integer.class);
        // 从响应中提取总页数，需根据实际接口返回的字段调整
        Integer totalPages = currentPage.getPageInfo().getTotalPages();
        // 聚合已拉取的所有页数据
        List<ResponseTesouroXXXXX> allPages = (List<ResponseTesouroXXXXX>) message.getHeaders().get("allPages");
        if (allPages == null) {
            allPages = new ArrayList<>();
        }
        allPages.add(currentPage);

        return MessageBuilder.withPayload(currentPage)
                .copyHeaders(message.getHeaders())
                .setHeader("currentPage", currentPageNum + 1)
                .setHeader("totalPages", totalPages)
                .setHeader("allPages", allPages)
                .build();
    }
}

3. 新增分页检查路由器实现循环调用 #

通过表达式路由器判断是否还有下一页需要拉取，有则路由回网关请求通道继续调用，无则进入最终聚合流程：

<int:router 
    input-channel="pageCheckRouterChannel" 
    expression="headers.currentPage <= headers.totalPages ? 'publishSubRequest' : 'finalAggregateChannel'"/>

4. 新增最终数据聚合节点 #

所有页拉取完成后，将全量数据聚合后发送到原有的分流处理流程：

<int:service-activator 
    input-channel="finalAggregateChannel" 
    output-channel="agora-consolidador-splitter-xxxxx-direto-xxxx" 
    ref="pageAggregateHandler" 
    method="aggregate"/>

对应Java处理类实现（可根据实际业务逻辑调整聚合规则）：

public class PageAggregateHandler {
    public List<ResponseTesouroXXXXX> aggregate(Message<?> message) {
        return (List<ResponseTesouroXXXXX>) message.getHeaders().get("allPages");
    }
}

可选优化点 #

• 可在循环路由路径中添加<int:delay>组件控制接口调用频率，避免触发限流
• 可搭配Spring Retry组件实现单页调用失败的重试逻辑
• 若分页数据量极大，可将中间页数据落地到临时存储，避免内存占用过高

内容的提问来源于stack exchange，提问作者Gabriel Gomes Crispim