有没有类似于迭代器的东西，但具有类似于Streams的函数

Question

所以基本上我要做的是：

1. 从数据库加载一批数据
2. 将该数据(Object[]查询结果(映射到以可读格式表示数据的类
3. 写入文件
4. 重复，直到查询不再得到结果为止

我列出了我熟悉的似乎符合需求的结构，以及它们不符合我需求的原因。

• 迭代器→没有在不调用next()的情况下映射和筛选的选项
  • 我需要在一个子类中定义map函数，尽管实际上没有数据(类似于流(，这样我就可以通过；流"；一直到一个调用类，并且只有在那里调用next，然后作为结果调用所有映射函数
• 流→在映射和筛选之前，所有数据都必须可用
• 可观察→一旦数据可用，就立即发送数据。不过我需要同步处理

为了更深入地了解我要做的事情，我举了一个小例子：

// Disclaimer: "Something" is the structure I am not sure of now. 
// Could be an Iterator or something else that fits (Thats the question)
public class Orchestrator {
@Inject
private DataGetter dataGetter;
public void doWork() {
FileWriter writer = new FileWriter("filename");
// Write the formatted data to the file
dataGetter.getData()
.forEach(data -> writer.writeToFile(data));
}
}
public class FileWriter {
public void writeToFile(List<Thing> data) {
// Write to file
}
}
public class DataGetter {
@Inject
private ThingDao thingDao;
public Something<List<Thing>> getData() {
// Map data to the correct format and return that
return thingDao.getThings()
.map(partialResult -> /* map to object */);
}
}
public class ThingDao {
public Something<List<Object[]>> getThings() {
Query q = ...;
// Dont know what to return
}
}

到目前为止我得到的：

我试着从迭代程序的基础开始，因为它是唯一一个真正满足我内存需求的迭代程序。然后我添加了一些方法来映射和循环数据。不过，这并不是一个真正稳健的设计，而且会比我想象的更难，所以我想知道是否有什么东西可以满足我的需求。

public class QIterator<E> implements Iterator<List<E>> {
public static String QUERY_OFFSET = "queryOffset";
public static String QUERY_LIMIT = "queryLimit";
private Query query;
private long lastResultIndex = 0;
private long batchSize;
private Function<List<Object>, List<E>> mapper;
public QIterator(Query query, long batchSize) {
this.query = query;
this.batchSize = batchSize;
}
public QIterator(Query query, long batchSize, Function<List<Object>, List<E>> mapper) {
this(query, batchSize);
this.mapper = mapper;
}
@Override
public boolean hasNext() {
return lastResultIndex % batchSize == 0;
}
@Override
public List<E> next() {
query.setParameter(QueryIterator.QUERY_OFFSET, lastResultIndex);
query.setParameter(QueryIterator.QUERY_LIMIT, batchSize);
List<Object> result = (List<Object>) query.getResultList(); // unchecked
lastResultIndex += result.size();
List<E> mappedResult;
if (mapper != null) {
mappedResult = mapper.apply(result);
} else {
mappedResult = (List<E>) result; // unchecked
}
return mappedResult;
}
public <R> QIterator<R> map(Function<List<E>, List<R>> appendingMapper) {
return new QIterator<>(query, batchSize, (data) -> {
if (this.mapper != null) {
return appendingMapper.apply(this.mapper.apply(data));
} else {
return appendingMapper.apply((List<E>) data);
}
});
}
public void forEach(BiConsumer<List<E>, Integer> consumer) {
for (int i = 0; this.hasNext(); i++) {
consumer.accept(this.next(), i);
}
}
}

到目前为止，这是有效的，但有一些unchecked任务，我真的不喜欢，我也希望有能力"；附加"；一个QIterator到另一个Qiderator，这本身并不难，但它也应该采用附加后的映射。

Answer 1

假设您有一个以分页方式提供数据的DAO，例如通过将LIMIT和OFFSET子句应用于底层SQL。这样的DAO类将有一个将这些值作为参数的方法，即该方法将符合以下函数方法：

@FunctionalInterface
public interface PagedDao<T> {
List<T> getData(int offset, int limit);
}

例如，调用getData(0, 20)将返回前20行(第1页(，而调用getData(60, 20)将返回第4页上的20行。如果该方法返回的行少于20行，则意味着我们得到了最后一页。在最后一行之后询问数据将返回一个空列表。

对于下面的演示，我们可以模拟这样一个DAO类：

public class MockDao {
private final int rowCount;
public MockDao(int rowCount) {
this.rowCount = rowCount;
}
public List<SimpleRow> getSimpleRows(int offset, int limit) {
System.out.println("DEBUG: getData(" + offset + ", " + limit + ")");
if (offset < 0 || limit <= 0)
throw new IllegalArgumentException();
List<SimpleRow> data = new ArrayList<>();
for (int i = 0, rowNo = offset + 1; i < limit && rowNo <= this.rowCount; i++, rowNo++)
data.add(new SimpleRow("Row #" + rowNo));
System.out.println("DEBUG:   data = " + data);
return data;
}
}
public class SimpleRow {
private final String data;
public SimpleRow(String data) {
this.data = data;
}
@Override
public String toString() {
return "Row[data=" + this.data + "]";
}
}

如果您想从该方法生成一个行的Stream，以特定大小的块流式传输所有行，我们需要一个Spliterator，因此我们可以使用StreamSupport.stream(Spliterator<T> spliterator, boolean parallel)来创建流。

以下是这样一个Spliterator:的实现

public class PagedDaoSpliterator<T> implements Spliterator<T> {
private final PagedDao<T> dao;
private final int blockSize;
private int nextOffset;
private List<T> data;
private int dataIdx;
public PagedDaoSpliterator(PagedDao<T> dao, int blockSize) {
if (blockSize <= 0)
throw new IllegalArgumentException();
this.dao = Objects.requireNonNull(dao);
this.blockSize = blockSize;
}
@Override
public boolean tryAdvance(Consumer<? super T> action) {
if (this.data == null) {
if (this.nextOffset == -1/*At end*/)
return false; // Already at end
this.data = this.dao.getData(this.nextOffset, this.blockSize);
this.dataIdx = 0;
if (this.data.size() < this.blockSize)
this.nextOffset = -1/*At end, after this data*/;
else
this.nextOffset += data.size();
if (this.data.isEmpty()) {
this.data = null;
return false; // At end
}
}
action.accept(this.data.get(this.dataIdx++));
if (this.dataIdx == this.data.size())
this.data = null;
return true;
}
@Override
public Spliterator<T> trySplit() {
return null; // Parallel processing not supported
}
@Override
public long estimateSize() {
return Long.MAX_VALUE; // Unknown
}
@Override
public int characteristics() {
return ORDERED | NONNULL;
}
}

我们现在可以使用上面的模拟DAO来测试它：

MockDao dao = new MockDao(13);
Stream<SimpleRow> stream = StreamSupport.stream(
new PagedDaoSpliterator<>(dao::getSimpleRows, 5), /*parallel*/false);
stream.forEach(System.out::println);

输出

DEBUG: getData(0, 5)
DEBUG:   data = [Row[data=Row #1], Row[data=Row #2], Row[data=Row #3], Row[data=Row #4], Row[data=Row #5]]
Row[data=Row #1]
Row[data=Row #2]
Row[data=Row #3]
Row[data=Row #4]
Row[data=Row #5]
DEBUG: getData(5, 5)
DEBUG:   data = [Row[data=Row #6], Row[data=Row #7], Row[data=Row #8], Row[data=Row #9], Row[data=Row #10]]
Row[data=Row #6]
Row[data=Row #7]
Row[data=Row #8]
Row[data=Row #9]
Row[data=Row #10]
DEBUG: getData(10, 5)
DEBUG:   data = [Row[data=Row #11], Row[data=Row #12], Row[data=Row #13]]
Row[data=Row #11]
Row[data=Row #12]
Row[data=Row #13]

可以看出，我们得到了13行数据，这些数据以5行为一个块从数据库中检索。

在需要数据之前，不会从数据库中检索数据，这会导致内存占用率低，这取决于块大小和不缓存数据的流操作。

Answer 2

您可以在一行中完成，如下所示：

stmt = con.createStatement();
ResultSet rs = stmt.executeQuery(queryThatReturnsAllRowsOrdered);
Stream.generate(rs.next() ? map(rs) : null)
.takeWhile(Objects::nonNull)
.filter(<some predicate>)
.forEach(<some operation);

当从查询返回第一行时，这将开始处理，并与数据库并行进行，直到读取完所有行。

这种方法一次只在内存中有一行，并且只运行一个查询就可以最大限度地减少数据库的负载。

从ResultSet进行映射要比从Object[]进行映射简单自然得多，因为您可以通过名称和正确键入的值来访问列，例如：

MyDao map(ResultSet rs) {
try {
String someStr = rs.getString("COLUMN_X");
int someInt = rs.getInt("COLUMN_Y"):
return new MyDao(someStr, someInt);
} catch (SQLException e ) {
throw new RuntimeException(e);
}
}