我在Java并行流中发现了一些令人惊讶的行为。我自己创建了一个Spliterator
,结果并行流被划分到每个流只包含一个元素。这似乎太小了,我想知道我做错了什么。我希望有一些特性可以设置来纠正这一点。
下面是我的测试代码。这里的Float
仅是一个虚拟载荷,我的真实流类要复杂得多。
public static void main( String[] args ) {
TestingSpliterator splits = new TestingSpliterator( 10 );
Stream<Float> test = StreamSupport.stream( splits, true );
double total = test.mapToDouble( Float::doubleValue ).sum();
System.out.println( "Total: " + total );
}
这段代码将不断分割流,直到每个Spliterator
只有一个元素。这似乎过于繁琐,不够高效。
输出:
run:
Split on count: 10
Split on count: 5
Split on count: 3
Split on count: 5
Split on count: 2
Split on count: 2
Split on count: 3
Split on count: 2
Split on count: 2
Total: 5.164293184876442
BUILD SUCCESSFUL (total time: 0 seconds)
这是
Spliterator
的代码。我的主要关注点是应该使用哪些特性,但也许还有其他问题吗?public class TestingSpliterator implements Spliterator<Float> {
int count;
int splits;
public TestingSpliterator( int count ) {
this.count = count;
}
@Override
public boolean tryAdvance( Consumer<? super Float> cnsmr ) {
if( count > 0 ) {
cnsmr.accept( (float)Math.random() );
count--;
return true;
} else
return false;
}
@Override
public Spliterator<Float> trySplit() {
System.err.println( "Split on count: " + count );
if( count > 1 ) {
splits++;
int half = count / 2;
TestingSpliterator newSplit = new TestingSpliterator( count - half );
count = half;
return newSplit;
} else
return null;
}
@Override
public long estimateSize() {
return count;
}
@Override
public int characteristics() {
return IMMUTABLE | SIZED;
}
}
那么如何将流分割成更大的块呢?我希望每个块的大小在10,000到50,000左右会更好。我知道可以从trySplit()方法中返回null,但那似乎是一种错误的做法。系统应该根据核心数、当前负载以及使用流的代码的复杂度等因素自行调整,从而外部配置流块大小,而不是由流本身内部固定。编辑:关于下面Holger的答案,当我增加原始流中的元素数量时,流分片会相对减少,所以StreamSupport最终会停止分片。在初始流大小为100个元素时,当达到2个流大小时(我屏幕上看到的最后一行是Split on count: 4),StreamSupport会停止分割。对于初始流大小为1000个元素的情况,每个单独的流块的最终大小约为32个元素。编辑第二部分:在查看以上输出之后,我更改了我的代码以列出创建的各个Spliterator。以下是更改内容:
public static void main( String[] args ) {
TestingSpliterator splits = new TestingSpliterator( 100 );
Stream<Float> test = StreamSupport.stream( splits, true );
double total = test.mapToDouble( Float::doubleValue ).sum();
System.out.println( "Total Spliterators: " + testers.size() );
for( TestingSpliterator t : testers ) {
System.out.println( "Splits: " + t.splits );
}
}
并且对于 TestingSpliterator
的构造函数:
static Queue<TestingSpliterator> testers = new ConcurrentLinkedQueue<>();
public TestingSpliterator( int count ) {
this.count = count;
testers.add( this ); // OUCH! 'this' escape
}
这段代码的结果是第一个
Spliterator
被分成5个部分。下一个Spliterator
被分成4个部分。接下来的一组Spliterators
被分成3个部分。以此类推。结果是创建了36个Spliterators
,流被分成相同数量的部分。在典型的桌面系统上,这似乎是API认为最适合并行操作的方式。
我将接受下面Holger的答案,基本上是StreamSupport
类正在做正确的事情,不要担心,开心点。对于我来说问题的一部分是我在非常小的流大小上进行早期测试,我对拆分的次数感到惊讶。不要犯同样的错误。
StreamSupport
将会分割成长度为2的流(我看到的最后一次分割是在“Split on count: 4”)。对于总长度为1000个元素的情况,分割将会降至大约32个元素。我将更新我的问题。 - markspaceStreamSupport
类没有问题,它的行为是合理的。感谢您指引我正确的方向。 - markspace