当Spring Batch中的数据是动态的时，如何用分区器重复作业？

Question

public class DocumentItemReader implements ItemReader<Document> {

    public List<Document> documents = new ArrayList<>();

    @Override
    public Document read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {

        if(documents.isEmpty()) {
            getDocuments(); // This method retrieve 100 documents and store them in "documents" list.
            if(documents.isEmpty()) return null;
        }

        Document doc = documents.get(0);
        documents.remove(0);
        return doc;
    }
}

但是，如果我想使用几个线程，问题就出现了。在本例中，我开始使用分区器方法而不是多线程。这样做的原因是因为我从同一个数据库读取，所以如果我用几个线程重复整个步骤，所有线程都将找到相同的记录，我不能使用分页（见下文）。

另一个问题是数据库记录是动态更新的，所以我不能使用分页。例如，让我们假设我有200条记录，并且所有记录都将很快过期，因此流程将检索它们。现在，假设我用一个线程检索了10个，在其他任何事情之前，这个线程处理了一个并在同一个数据库中更新它。下一个线程不能检索11到20条记录，因为第一条记录不会出现在搜索中（因为它已经被处理，它的日期已经更新，然后它与查询不匹配）。

有点难以理解，有些事情听起来可能很奇怪，但在我的项目中：

null

谢谢你。

Answer 1

也许您可以在您的步骤中添加一个分区器，该分区器将：

1. 选择需要更新的数据的所有ID（如果需要，请选择其他列）
2. 将它们拆分到x个(x=gridSize参数）分区中，并将它们写入临时文件（按分区1）。
3. 在ExecutionContext中注册要读取的文件名

那么您的阅读器将不再从数据库中读取，而是从分区文件中读取。

public class JdbcToFilePartitioner implements Partitioner {

    /** number of records by database fetch  */
    private int fetchSize = 100;

    /** working directory */
    private File tmpDir;

    /** limit the number of item to select */
    private Long nbItemMax;

    @Override
    public Map<String, ExecutionContext> partition(final int gridSize) {

        // Create contexts for each parttion
        Map<String, ExecutionContext> executionsContexte = createExecutionsContext(gridSize);

        // Fill partition with ids to handle
        getIdsAndFillPartitionFiles(executionsContexte);

        return executionsContexte;
    }

    /**
     * @param gridSize number of partitions
     * @return map of execution context, one for each partition
     */
    private Map<String, ExecutionContext> createExecutionsContext(final int gridSize) {

        final Map<String, ExecutionContext> map = new HashMap<>();

        for (int partitionId = 0; partitionId < gridSize; partitionId++) {
            map.put(String.valueOf(partitionId), createContext(partitionId));
        }

        return map;
    }

    /**
     * @param partitionId id of the partition to create context
     * @return created executionContext
     */
    private ExecutionContext createContext(final int partitionId) {

        final ExecutionContext context = new ExecutionContext();

        String fileName = tmpDir + File.separator + "partition_" + partitionId + ".txt";

        context.put(PartitionerConstantes.ID_GRID.getCode(), partitionId);
        context.put(PartitionerConstantes.FILE_NAME.getCode(), fileName);

        if (contextParameters != null) {
            for (Entry<String, Object> entry : contextParameters.entrySet()) {
                context.put(entry.getKey(), entry.getValue());
            }
        }

        return context;
    }

    private void getIdsAndFillPartitionFiles(final Map<String, ExecutionContext> executionsContexte) {

        List<BufferedWriter> fileWriters = new ArrayList<>();
        try {

            // BufferedWriter for each partition
            for (int i = 0; i < executionsContexte.size(); i++) {
                BufferedWriter bufferedWriter = new BufferedWriter(new FileWriter(executionsContexte.get(String.valueOf(i)).getString(
                        PartitionerConstantes.FILE_NAME.getCode())));
                fileWriters.add(bufferedWriter);
            }

            // Fetching the datas
            ScrollableResults results = runQuery();

            // Get the result and fill the files
            int currentPartition = 0;
            int nbWriting = 0;
            while (results.next()) {
                fileWriters.get(currentPartition).write(results.get(0).toString());
                fileWriters.get(currentPartition).newLine();
                currentPartition++;
                nbWriting++;

                // If we already write on all partitions, we start again
                if (currentPartition >= executionsContexte.size()) {
                    currentPartition = 0;
                }

                // If we reach the max item to read we stop
                if (nbItemMax != null && nbItemMax != 0 && nbWriting >= nbItemMax) {
                    break;
                }
            }

            // closing
            results.close();
            session.close();
            for (BufferedWriter bufferedWriter : fileWriters) {
                bufferedWriter.close();
            }
        } catch (IOException | SQLException e) {
            throw new UnexpectedJobExecutionException("Error writing partition file", e);
        }
    }

    private ScrollableResults runQuery() {
        ...
    }
}