我们进行了类似的练习，将多个表中的100mn+行作为json的一种形式进行迁移，以便将其发布到消息总线。
其思想是创建一个视图，对数据进行非规范化，并使用jdbcpagingitemreader从该视图中读取数据。从一个源读取数据的开销较小。
反规范化数据时，请确保主表中没有多行。
示例-sql server-

create or alter view viewName as
select master.col1 , master.col2,
(select dep1.col1,
               dep1.col2
               from dependent1 dep1
        where dep1.col3 = master.col3 for json path
       )                as dep1
from master master;

上面的内容将为您提供json字符串中的依赖表数据，每个主表数据对应一行。一旦检索到数据，就可以使用gson或jackson将其转换为pojo。
我们试图避免使用jdbccursoritemreader，因为它将从内存中提取所有数据，并从中逐个读取数据。它不支持分页。

我们在一个项目中有这样一个需求，并用以下方法解决了它。
在并行运行的splitflow中，我们有一个step for ever表，它将表的数据加载到文件中，并按公共id排序（这是可选的，但如果文件中有数据，则测试更容易）。
然后我们实现了自己的“mergereader”。这个mergereader对每个文件/表都有flatfileitemreaders（我们称之为datareaders）。所有这些flatfileitemreaders都用一个singleitempeakeableitemreader Package 。mergereader的read方法的逻辑如下：

public MyContainerPerId read() {

   // you need a container to store the items, that belong together
   MyContainerPerId container = new MyContainerPerId();

   //  peek through all "dataReaders" to find the lowest actual key
   int lowestId = searchLowestKey();

   for (Reader dataReader : dataReaders) {
       // I assume, that more than one entry in a table can belong to
       // the same person id
       wihile (dataReader.peek().getId() == lowestId) {
       {
             container.add(dataReader.read());
       }
   }

   // the container contains all entries from all tables
   // belonging to the same person id    
   return container;
}

如果您需要重新启动功能，您可以以某种方式实现itemstream，它可以跟踪每个datareader的当前readposition。

我使用这里描述的基于驱动查询的itemreaders使用模式来解决这个问题。
reader：只是jdbccursoritemreader的一个默认实现，使用sql获取
唯一的关系id（例如，从person-中选择id）
processor：使用这个长id作为输入，我使用spring的jdbctemplate实现的dao通过查询每个表获取特定id的数据（例如，select*from license where id=），并将结果以列表格式Map到person的pojo，然后转换为json对象（使用jackson），然后转换为字符串
writer：如果使用kafka，可以用json字符串写出文件，也可以将json字符串发布到主题中