Spring Integration Java DSL：如何循环分页的 Rest 服务？

Question

如何使用 Java DSL Http.outboundGatewaymethod 循环分页的 Rest 服务？

其余网址例如

http://localhost:8080/people?page=3

它返回例如

"content": [
    {"name": "Mike",
     "city": "MyCity"
    },
    {"name": "Peter",
     "city": "MyCity"
    },
    ...
 ]
"pageable": {
    "sort": {
        "sorted": false,
        "unsorted": true
    },
    "pageSize": 20,
    "pageNumber": 3,
    "offset": 60,
    "paged": true,
    "unpaged": false
},
"last": false,
"totalElements": 250,
"totalPages": 13,
"first": false,
"sort": {
    "sorted": false,
    "unsorted": true
},
"number": 3,
"numberOfElements": 20,
"size": 20
}

变量totalPages 表示总页数。

所以如果实现

        integrationFlowBuilder
          .handle(Http
            .outboundGateway("http://localhost:8080/people?page=3")
            .httpMethod(HttpMethod.GET)
            .expectedResponseType(String.class))

访问一个页面，如何循环所有页面？

Answer 1

最简单的方法是使用@MessagingGateway 包装对Http.outboundGateway() 的调用，并提供页码作为参数：

@MessagingGateway
public interface HttpPagingGateway {

    @Gateway(requestChannel = "httpPagingGatewayChannel")
    String getPage(int page);

}

然后你会得到一个 JSON 作为结果，你可以将它转换成一些域模型或者只是执行一个JsonPathUtils.evaluate()（基于json-path）来获取last属性的值，以确保你是否需要将getPage() 称为page++。

page 参数将是要发送的消息的payload，可以用作uriVariable：

.handle(Http
        .outboundGateway("http://localhost:8080/people?page={page}")
        .httpMethod(HttpMethod.GET)
        .uriVariable("page", Message::getPayload)
        .expectedResponseType(String.class))

当然，我们可以用 Spring Integration 做类似的事情，但会涉及到filter、router 和其他一些组件。

更新

首先，我建议您创建一个域模型（一些 Java Bean），例如 PersonPageResult，以将 JSON 响应和此类型表示为 Http.outboundGateway() 的 expectedResponseType(PersonPageResult.class) 属性。开箱即用的RestTemplate 和MappingJackson2HttpMessageConverter 可以帮助您返回这样的对象作为下游处理的回复。

然后，正如我之前所说，循环最好从一些 Java 代码中完成，您可以将其包装到服务激活器调用中。为此，您应该像这样 daclare 一个网关：

public interface HttpPagingGateway {

    PersonPageResult getPage(int page);

}

注意：根本没有注释。这个技巧是通过IntegrationFlow完成的：

@Bean
public IntegrationFlow httpGatewayFlow() {
    return IntegrationFlows.from(HttpPagingGateway.class)
                  .handle(Http
                       .outboundGateway("http://localhost:8080/people?page={page}")
                       .httpMethod(HttpMethod.GET)
                       .uriVariable("page", Message::getPayload)
                       .expectedResponseType(PersonPageResult.class))  
}

参见IntegrationFlows.from(Class<?> aClass)JavaDocs。

这样的HttpPagingGateway 可以通过硬循环逻辑注入到某些服务中：

int page = 1;
boolean last = false;
while(!last) {
  PersonPageResult result = this.httpPagingGateway.getPage(page++);
  last = result.getLast();
  List<Person> persons = result.getPersons();
  // Process persons
}

为了处理那些persons，我建议有单独的IntegrationFlow，它也可以从网关开始，或者你可以发送一个Message<List<Person>>到它的输入通道。

这样你就可以将分页和处理的关注点分开，并且在一些 POJO 方法中会有一个简单的循环逻辑。