迁移到 Spring Boot 2 并使用 Spring Batch 4

Question

我正在将 Spring Boot 从 1.4.2 迁移到 2.0.0，其中还包括将 Spring 批处理从 3.0.7 迁移到 4.0.0，当我尝试使用新 Spring 运行批处理时，它看起来不再有效批量版本。

当我尝试调试时，我发现批处理尝试从 batch_job_execution_context 获取数据时出现问题。

我可以看到从数据库中获取数据工作正常，但是新版本的批处理无法解析数据库数据

{"map":[{"entry":[{"string":["name",""]},{"string":["sender",""]},{"string": ["id",""]},{"string":["nav",""]},{"string":["created",140418]}]}]}

出现此错误：

com.fasterxml.jackson.databind.exc.MismatchedInputException: Unexpected token (START_OBJECT), expected VALUE_STRING: need JSON String that contains type id (for subtype of java.lang.Object) at [Source: (ByteArrayInputStream); line: 1, column: 9] (through reference chain: java.util.HashMap["map"])

我发现，当我删除所有批处理元数据表并从头开始重新创建它们时，批处理似乎又可以工作了。元数据 JSON 格式好像变成了这个

{"name":"","sender":"145844","id":"","nav":"","created":"160909"}

我不想删除旧数据以使其再次工作，那么有什么办法可以解决这个问题吗？

还有其他人尝试过进行此升级吗？很高兴知道是否还有其他我可能没有注意到的重大变化。

谢谢

Answer 1

在 Spring Batch 4 之前，ExecutionContext 的默认序列化机制是通过 XStream。现在它默认使用 Jackson，它与旧的序列化格式不兼容。我们仍有可用的旧版本 (XStreamExecutionContextStringSerializer)，但您需要通过实现 BatchConfigurer 并覆盖 JobRepositoryFactoryBean 中的配置来自行配置。

记录在案，这与此问题有关：https://jira.spring.io/browse/BATCH-2575。

Answer 2

根据上面迈克尔的回答，这个代码块对我来说是为了扩展默认配置——我必须将 Serializer 连接到 JobRepository.class 和 JobExplorer.class：

@Configuration
@EnableBatchProcessing
MyBatchConfigurer extends DefaultBatchConfigurer {
    private final DataSource dataSource;

    @Autowired
    public BatchConfiguration(final DataSource dataSource) throws Exception {
        this.dataSource = dataSource;
    }

    @Bean
    ExecutionContextSerializer getSerializer() {
        return new XStreamExecutionContextStringSerializer();
    }


    @Override
    protected JobRepository createJobRepository() throws Exception {
        final JobRepositoryFactoryBean factory = new JobRepositoryFactoryBean();
        factory.setDataSource(dataSource);
        factory.setSerializer(getSerializer());
        factory.setTransactionManager(getTransactionManager());
        factory.afterPropertiesSet();
        return factory.getObject();
    }

    @Override
    protected JobExplorer createJobExplorer() throws Exception {
        final JobExplorerFactoryBean jobExplorerFactoryBean = new JobExplorerFactoryBean();
        jobExplorerFactoryBean.setDataSource(dataSource);
        jobExplorerFactoryBean.setSerializer(getSerializer());
        jobExplorerFactoryBean.afterPropertiesSet();
        return jobExplorerFactoryBean.getObject();
    }
}

Answer 3

在@anotherdave 和@michael-minella 的解决方案中，您还可以将普通的XStreamExecutionContextStringSerializer 替换为以下类的实例。它在反序列化和序列化为新格式时接受这两种格式。

import java.io.BufferedInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.Map;
import com.fasterxml.jackson.core.JsonProcessingException;
import org.springframework.batch.core.repository.ExecutionContextSerializer;
import org.springframework.batch.core.repository.dao.Jackson2ExecutionContextStringSerializer;
import org.springframework.batch.core.repository.dao.XStreamExecutionContextStringSerializer;


/**
 * Enables Spring Batch 4 to read both ExecutionContext entries written by ealier versions and the Spring 5 format. Entries are
 * written in Spring 5 format.
 */
@SuppressWarnings("deprecation")
class XStreamOrJackson2ExecutionContextSerializer implements ExecutionContextSerializer {
    private final XStreamExecutionContextStringSerializer xStream = new XStreamExecutionContextStringSerializer();
    private final Jackson2ExecutionContextStringSerializer jackson = new Jackson2ExecutionContextStringSerializer();

    public XStreamOrJackson2ExecutionContextSerializer() throws Exception {
        xStream.afterPropertiesSet();
    }

    // The caller closes the stream; and the decoration by ensureMarkSupported does not need any cleanup.
    @SuppressWarnings("resource")
    @Override
    public Map<String, Object> deserialize(InputStream inputStream) throws IOException {
        InputStream repeatableInputStream = ensureMarkSupported(inputStream);
        repeatableInputStream.mark(Integer.MAX_VALUE);

        try {
            return jackson.deserialize(repeatableInputStream);
        } catch (JsonProcessingException e) {
            repeatableInputStream.reset();
            return xStream.deserialize(repeatableInputStream);
        }
    }

    private static InputStream ensureMarkSupported(InputStream in) {
        return in.markSupported() ? in : new BufferedInputStream(in);
    }

    @Override
    public void serialize(Map<String, Object> object, OutputStream outputStream) throws IOException {
        jackson.serialize(object, outputStream);
    }
}