春季批处理以退出状态退出:在实际作业完成之前已完成? [英] Spring-batch exiting with Exit Status : COMPLETED before actual job is finished?

问题描述

在我的Spring Batch应用程序中，我编写了一个CustomItemWriter，它使用DynamoDBAsyncClient在内部将项目写入DynamoDB，此客户端返回Future对象.我有一个包含数百万条记录的输入文件.由于CustomItemWriter立即返回将来的对象，所以我的批处理作业将在5秒钟内退出，状态为COMPLETED，但是实际上将所有项写入数据库需要3-4分钟，因此我希望该批处理作业仅在将所有项写入DataBase之后才能完成.我该怎么办?

In my Spring Batch application I have written a CustomItemWriter which internally writes item to DynamoDB using DynamoDBAsyncClient, this client returns Future object. I have a input file with millions of record. Since CustomItemWriter returns future object immediately my batch job exiting within 5 sec with status as COMPLETED, but in actual it is taking 3-4 minutes to write all item to the DB, I want that batch job finishes only after all item written to DataBase. How can i do that?

工作定义如下

    <bean id="report" class="com.solution.model.Report" scope="prototype" />
        <batch:job id="job" restartable="true">
            <batch:step id="step1">
                <batch:tasklet>
                    <batch:chunk reader="cvsFileItemReader"  processor="filterReportProcessor" writer="customItemWriter"
                        commit-interval="20">
                    </batch:chunk>
                </batch:tasklet>
            </batch:step>
        </batch:job>
<bean id="customItemWriter" class="com.solution.writer.CustomeWriter"></bean>

CustomeItemWriter定义如下

CustomeItemWriter is defined as below

public class CustomeWriter implements ItemWriter<Report>{
    public void write(List<? extends Report> item) throws Exception {
    List<Future<PutItemResult>> list = new LinkedList();
    AmazonDynamoDBAsyncClient client = new AmazonDynamoDBAsyncClient();
        for(Report report : item) {
            PutItemRequest req = new PutItemRequest();
            req.setTableName("MyTable");
            req.setReturnValue(ReturnValue.ALL_ODD);
            req.addItemEntry("customerId",new 
            AttributeValue(item.getCustomeId()));
            Future<PutItemResult> res = client.putItemAsync(req);
            list.add(res);
            }
    }

}

主类包含

JobExecution execution = jobLauncher.run(job, new JobParameters());
System.out.println("Exit Status : " + execution.getStatus());

由于在ItemWriter中返回了将来的对象，因此它不等待完成操作.从主要方面来说，由于所有项目均已提交以进行写入，因此批次状态显示为COMPLETED，并且作业终止.我希望仅在DynamoDB中执行实际写操作后才能终止此作业.我们可以再等一步吗?还是可以使用某些侦听器?

Since in ItemWriter its returning future object it doesn't waits to complete the opration. And from the main since all item is submitted for writing Batch Status is showing COMPLETED and job terminates. I want that this job should terminate only after actual write is performed in the DynamoDB. Can we have some other step well to wait on this or some Listener is available?

推荐答案

这是一种方法.由于 ItemWriter :: write 不返回任何内容，因此您可以使用侦听器功能.

Here is one approach. Since ItemWriter::write doesn't return anything you can make use of listener feature.

@Component
@JobScope
public class YourWriteListener implements ItemWriteListener<WhatEverYourTypeIs> {


  @Value("#{jobExecution.executionContext}")
  private ExecutionContext executionContext;


  @Override
  public void afterWrite(final List<? extends WhatEverYourTypeIs> paramList) {
     Future future = this.executionContext.readAndValidate("FutureKey", Future.class);
     //wait till the job is done using future object
  }

  @Override
  public void beforeWrite(final List<? extends WhatEverYourTypeIs> paramList) {

  }

  @Override
  public void onWriteError(final Exception paramException, final List<? extends WhatEverYourTypeIs> paramList) {

  }
}

在您的writer类中，除了将将来的对象添加到ExecutionContext外，其他所有内容均保持不变.

In your writer class, everything remains same except addind the future object to ExecutionContext.

public class YourItemWriter extends ItemWriter<WhatEverYourTypeIs> {

  @Value("#{jobExecution.executionContext}")
  private ExecutionContext executionContext;

  @Override
  protected void doWrite(final List<? extends WhatEverYourTypeIs> youritems) 

     //write to DynamoDb and get Future object
    executionContext.put("FutureKey", future);
    }

  }

}

您可以在配置中注册侦听器.这是一个Java代码，您需要在xml中做同样的事情

And you can register the listener in your configuration. Here is a java code, you need to do the same in your xml

@Bean
  public Step initStep() {

    return this.stepBuilders.get("someStepName").<YourTypeX, YourTypeY>chunk(10)
        .reader(yourReader).processor(yourProcessor)
        .writer(yourWriter).listener(YourWriteListener)
        .build();
  }

这篇关于春季批处理以退出状态退出:在实际作业完成之前已完成?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！