带有持久存储作业注册表的 spring batch 重新启 Action 业

Question

我正在尝试在我的 spring boot 应用程序中使用 spring-batch。我有一个用例在一天中的特定时间段内停止作业，一旦时间窗口过去，我想从暂停的步骤重新启 Action 业。到目前为止，我已经发现有两种方法可以做到这一点，如下所示:

class JobStarter {
  @Autowired protected JobLauncher jobLauncher
  @Autowired protected JobRegistry jobRegistry
  @Autowired protected JobOperator jobOperator
  @Autowired protected JobExplorer jobExplorer

  def startJob(Map config) {
    Job job = build(config, subject) // build the job programmatically
    JobParameters jobParameters = createJobParameters(subject, config) 
    jobLauncher.launch (job, jobParameters)
    jobRegistry.register(new ReferenceFactory(job))
  }

  def restartJob1(String jobName, JobParameters jobParameters) {
    Job job = jobRegistry.getJob(jobName)
    jobLauncher.launch (job, jobParameters)
  }

  // OR

  def restartJob2(String jobName) {
    def jobInstances = jobExplorer.getJobInstances(jobName, 0, 1)
    if (!jobInstances) {
      throw new Exception("Unable to find the job to restart")
    }
    def jobInstance = jobInstances[0]
    jobOperator.restart(jobInstance.instanceId)
  }
}

我有 jobRepository 和 jobExplorer 的自定义实现，它们都由持久存储支持。

我无法使用 spring batch 提供的 JobRegistry 的默认实现，因为它由内存中的 Map 支持，如果服务器出现故障，jobRegistry 为空，因此暂停作业的重启失败。我正在尝试实现 JobRegistry，但问题是我很难在我的 jobRegistry 自定义实现中坚持工作工厂:

class PersistentJobRegistry implements JobRegistry {

  // Other methods and code omitted for brevity
  static ObjectMapper mapper
  static {
    mapper = new ObjectMapper() 
    mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
    mapper.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false)
    mapper.enableDefaultTyping()
  }

  @Override
  void register(JobFactory jobFactory) throws DuplicateJobException {
    String jobFactoryStr = mapper.writeValueAsString(jobFactory)
    save(jobFactory.getJobName(), jobFactoryStr) // saves jobFactoryStr to store
  }

  @Override
  Job getJob(String name) throws NoSuchJobException {
    String jobFactoryStr = get(name) // gets a serialized factory from store
    JobFactory jobFactory = mapper.readValue(jobFactoryStr, JobFactory.class)
    return jobFactory.createJob()
  }
}

保存时，我得到的是 jobFactoryStr，这显然不起作用，因为只有 jobName 而没有关于作业及其步骤的其他信息:

jobFactoryStr = {"jobName":"MyJob:f6892e89-4a76-4395-ba0a-ca82b477a407"}

我在 getJob() 中也遇到以下错误:

com.fasterxml.jackson.databind.JsonMappingException: Unexpected token (START_OBJECT), expected START_ARRAY: need JSON Array to contain As.WRAPPER_ARRAY type information for class org.springframework.batch.core.configuration.JobFactory
 at [Source: {"jobName":"Pipeline:api:testing:c3241e19-4f32-474f-90d2-2974b672c183"}; line: 1, column: 1]
    at com.fasterxml.jackson.databind.JsonMappingException.from(JsonMappingException.java:148)
    at com.fasterxml.jackson.databind.DeserializationContext.wrongTokenException(DeserializationContext.java:854)
    at com.fasterxml.jackson.databind.jsontype.impl.AsArrayTypeDeserializer._locateTypeId(AsArrayTypeDeserializer.java:122)
    at com.fasterxml.jackson.databind.jsontype.impl.AsArrayTypeDeserializer._deserialize(AsArrayTypeDeserializer.java:93)
    at com.fasterxml.jackson.databind.jsontype.impl.AsArrayTypeDeserializer.deserializeTypedFromObject(AsArrayTypeDeserializer.java:58)
    at com.fasterxml.jackson.databind.deser.AbstractDeserializer.deserializeWithType(AbstractDeserializer.java:132)
    at com.fasterxml.jackson.databind.deser.impl.TypeWrappedDeserializer.deserialize(TypeWrappedDeserializer.java:41)
    at com.fasterxml.jackson.databind.ObjectMapper._readMapAndClose(ObjectMapper.java:3066)
    at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:2161)

我有两个问题:

1. 我重启作业的方法是否正确？

2. 我的 jobRegistry 实现方法是否正确？上述 Jackson 错误可能出了什么问题？

Answer 1

内置的 Job Repository 可以由任何 rdbms 支持，你不需要自己创建。默认情况下它被配置为指向内存数据库，但这是一个改变配置的问题属性文件。

从 job-repository 标记开始，通过数据源定义向后挖掘 spring 上下文和属性文件。将数据源指向目标数据库。

用于为最常用的 RDBMS 创建表的 DDL 嵌入在 jar 中，通常 spring-batch 会在第一次启动 spring-batch 时负责创建表(无论如何在你的 DEV 环境中)工作。

参见 http://docs.spring.io/spring-batch/reference/html/configureJob.html