我正在尝试使用Java8中的流创建一个生产者多个消费者模型。我正在从数据库资源中读取和处理数据,并希望以流式方式处理它们(不能将整个资源读入内存)。
源的读取必须是单线程的(游标不是线程安全的),而读取速度很快,因此每个数据块的处理操作可以并行运行。
我还没有找到如何连接非并行流和并行流处理的方法。是否有任何方法可以使用Java8流API实现?
代码示例:
这个迭代器必须在单线程中运行,因为游标不是线程安全的。
源的读取必须是单线程的(游标不是线程安全的),而读取速度很快,因此每个数据块的处理操作可以并行运行。
我还没有找到如何连接非并行流和并行流处理的方法。是否有任何方法可以使用Java8流API实现?
代码示例:
这个迭代器必须在单线程中运行,因为游标不是线程安全的。
class SimpleIterator<Data> implements Iterator<Data>{
private volatile Cursor cursor;
public SimpleIterator(Cursor cursor){
this.cursor = cursor;
}
@Override
public boolean hasNext() {
return cursor.hasNext();
}
@Override
public Data next() {
return cursor.next();
}
}
//创建非并行流
SimpleIterator<Data> iterator = new SimpleIterator<>(queryCursor);
Iterable<Data> iterable = () -> iterator;
Stream<Data> resultStream = StreamSupport.stream(iterable.spliterator(), false); // prallel set as false
//每个数据的处理都应该并行运行
resultStream.parallel().forEach(data->processData(data));
public processData(Data data){
//heavy operation
}
但是如果在调用forEach之前将流设置为并行,则整个流都是并行的,并且迭代器会在多个线程中调用。 在Java8中,是否有任何方法可以将这两个流连接起来,或者我必须创建某个队列,以从单线程生产者流提供数据到并行流中。