从 spring 批处理器调用异步 REST api

Question

我写了一个处理列表列表的春季批处理作业。

读取器返回列表列表。 处理器处理每个列表项并返回已处理的列表。 编写器从列表列表向数据库和 sftp 写入内容。

我有一个用例，我从弹簧批处理器调用异步 REST api。 在ListenableFuture响应中，我实现了LitenableFutureCallback来处理成功和失败，它按预期工作，但是在异步调用返回某些内容之前，ItemProcessor不会等待来自异步api的回调并将object(List(返回给编写器。

我不确定如何实现和处理来自项目处理器的异步调用。

我确实读过有关AsyncItemProcessor和AsyncItemWriter的信息，但我不确定在这种情况下是否应该使用。

我还想到从 AsyncRestTemplate 调用 ListenableFuture 响应的 get((，但根据文档，它将阻止当前线程，直到它收到响应。

我正在寻求有关如何实现这一点的帮助。下面的代码片段：

处理器：

public class MailDocumentProcessor implements ItemProcessor<List<MailingDocsEntity>, List<MailingDocsEntity>> {
... Initialization code
@Override
public List<MailingDocsEntity> process(List<MailingDocsEntity> documentsList) throws Exception {
logger.info("Entering MailingDocsEntity processor");

List<MailingDocsEntity> synchronizedList = Collections.synchronizedList(documentsList);

for (MailingDocsEntity mailingDocsEntity : synchronizedList) {
System.out.println("Reading Mailing id: " + mailingDocsEntity.getMailingId());
..code to get the file
//If the file is not a pdf convert it
String fileExtension = readFromSpResponse.getFileExtension();
String fileName = readFromSpResponse.getFileName();
byte[] fileBytes = readFromSpResponse.getByteArray();
try {
//Do checks to make sure PDF file is being sent
if (!"pdf".equalsIgnoreCase(fileExtension)) {
//Only doc, docx and xlsx conversions are supported
...Building REquest object
//make async call to pdf conversion service
pdfService.convertDocxToPdf(request, mailingDocsEntity);
} else {
logger.error("The file cannot be converted to a pdf.n"
);
}
}

} catch (Exception ex){
logger.error("There has been an exception while processing data", ex);
}
}
return synchronizedList;
}
}

异步 pdfConversion Service Class：

@Service
public class PdfService{

@Autowired
@Qualifier("MicroServiceAsyncRestTemplate")
AsyncRestTemplate microServiceAsyncRestTemplate;
public ConvertDocxToPdfResponse convertDocxToPdf(ConvertDocxToPdfRequest request, MailingDocsEntity mailingDocsEntity){
ConvertDocxToPdfResponse pdfResponse = new ConvertDocxToPdfResponse();

try {
HttpHeaders headers = new HttpHeaders();
headers.setContentType(MediaType.APPLICATION_JSON);
HttpEntity<?> entity = new HttpEntity<>(request, headers);

ListenableFuture<ResponseEntity<ConvertDocxToPdfResponse>> microServiceResponse = microServiceAsyncRestTemplate.postForEntity(batchMailProcessingConfiguration.getPdfUrl(), entity, ConvertDocxToPdfResponse.class);
ConvertDocxToPdfResponse resultBody = microServiceResponse.get().getBody();
microServiceResponse.addCallback(new ListenableFutureCallback<ResponseEntity<ConvertDocxToPdfResponse>>()  {
@Override
public void onSuccess(ResponseEntity<ConvertDocxToPdfResponse> result) {
...code to do stuff on success

}
@Override
public void onFailure(Throwable ex) {
pdfResponse.setMessage("Exception while retrieving response");
}
});
} catch (Exception e) {
String message = "There has been an error while issuing a pdf generate request to the pdf micro service";
pdfResponse.setMessage(message);
logger.error(message, e);
}

return pdfResponse;
}
}

我原来的批处理作业是同步的，我正在转换为异步以加快处理速度。 我确实试图寻找类似的问题，但找不到足够的信息。 任何指示或帮助都非常感谢。

谢谢！！

Answer 1

我确实读过有关AsyncItemProcessor和AsyncItemWriter的信息，但我不确定在这种情况下是否应该使用。

是的，AsyncItemProcessor和AsyncItemWriter适合您的用例。AsyncItemProcessor将为新线程上的项目执行委托ItemProcessor的逻辑(您的休息调用(。项目完成后，结果的Future将传递到要写入的AsynchItemWriter。然后，AsynchItemWriter将解开Future包装并写入项目。这些组件的优点是你不必自己处理Future包装、解包等。

您可以找到：

• 更多详情请见：https://docs.spring.io/spring-batch/4.0.x/reference/html/spring-batch-integration.html#asynchronous-processors
• 这里的例子：https://github.com/mminella/scaling-demos/blob/master/single-jvm-demos/src/main/java/io/spring/batch/scalingdemos/asyncprocessor/AsyncProcessorJobApplication.java

希望这有帮助。