我想使用 aStream来并行处理一组异构的远程存储的未知数量的 JSON 文件（文件的数量是未知的）。这些文件的大小差异很大，从每个文件 1 条 JSON 记录到其他一些文件中的 100,000 条记录。在这种情况下， JSON 记录意味着一个自包含的 JSON 对象，表示为文件中的一行。
我真的很想为此使用 Streams，所以我实现了这个Spliterator：
public abstract class JsonStreamSpliterator<METADATA, RECORD> extends AbstractSpliterator<RECORD> {
abstract protected JsonStreamSupport<METADATA> openInputStream(String path);
abstract protected RECORD parse(METADATA metadata, Map<String, Object> json);
private static final int ADDITIONAL_CHARACTERISTICS = Spliterator.IMMUTABLE | Spliterator.DISTINCT | Spliterator.NONNULL;
private static final int MAX_BUFFER = 100;
private final Iterator<String> paths;
private JsonStreamSupport<METADATA> reader = null;
public JsonStreamSpliterator(Iterator<String> paths) {
this(Long.MAX_VALUE, ADDITIONAL_CHARACTERISTICS, paths);
}
private JsonStreamSpliterator(long est, int additionalCharacteristics, Iterator<String> paths) {
super(est, additionalCharacteristics);
this.paths = paths;
}
private JsonStreamSpliterator(long est, int additionalCharacteristics, Iterator<String> paths, String nextPath) {
this(est, additionalCharacteristics, paths);
open(nextPath);
}
@Override
public boolean tryAdvance(Consumer<? super RECORD> action) {
if(reader == null) {
String path = takeNextPath();
if(path != null) {
open(path);
}
else {
return false;
}
}
Map<String, Object> json = reader.readJsonLine();
if(json != null) {
RECORD item = parse(reader.getMetadata(), json);
action.accept(item);
return true;
}
else {
reader.close();
reader = null;
return tryAdvance(action);
}
}
private void open(String path) {
reader = openInputStream(path);
}
private String takeNextPath() {
synchronized(paths) {
if(paths.hasNext()) {
return paths.next();
}
}
return null;
}
@Override
public Spliterator<RECORD> trySplit() {
String nextPath = takeNextPath();
if(nextPath != null) {
return new JsonStreamSpliterator<METADATA,RECORD>(Long.MAX_VALUE, ADDITIONAL_CHARACTERISTICS, paths, nextPath) {
@Override
protected JsonStreamSupport<METADATA> openInputStream(String path) {
return JsonStreamSpliterator.this.openInputStream(path);
}
@Override
protected RECORD parse(METADATA metaData, Map<String,Object> json) {
return JsonStreamSpliterator.this.parse(metaData, json);
}
};
}
else {
List<RECORD> records = new ArrayList<RECORD>();
while(tryAdvance(records::add) && records.size() < MAX_BUFFER) {
// loop
}
if(records.size() != 0) {
return records.spliterator();
}
else {
return null;
}
}
}
}
我遇到的问题是，虽然 Stream 一开始可以很好地并行化，但最终最大的文件会在单个线程中处理。我相信近端原因有据可查：分离器是“不平衡的”。
更具体地说，似乎trySplit在 的生命周期中的某个时间点之后不会调用该方法Stream.forEach，因此在末尾分发小批量的额外逻辑trySplit很少执行。
请注意从 trySplit 返回的所有拆分器如何共享相同的paths迭代器。我认为这是一种非常聪明的方法来平衡所有拆分器的工作，但它还不足以实现完全并行。
我希望首先跨文件进行并行处理，然后当仍然有少数大文件处于拆分状态时，我想跨剩余文件的块进行并行处理。那是else块末尾的意图trySplit。
是否有解决此问题的简单/简单/规范的方法？