java - 是否可以使用并行流来执行这些循环？

Question

如果我有在嵌套循环中执行的代码

int height = ...;
int width = ...;
int sourceIncrement = ...;
int destIncrement = ...;
int sourceOffset = ...;
int destOffset = ...;
int  sourceArray[] = ...;
byte destArray[] = ...;

for (int i = 0; i < height; i++) {                              
    for (int j = 0; j < width; j++) {               
        int pixel = sourceArray[sourceOffset ++];           
        destArray[destOffset ++] = (byte) (pixel      );
        destArray[destOffset ++] = (byte) (pixel >>  8);
        destArray[destOffset ++] = (byte) (pixel >> 16);
        destArray[destOffset ++] = (byte) (pixel >> 24);
    }                                           
    sourceOffset += sourceIncrement;                      
    destOffset += destIncrement;                     
}                                 

我想让它并行运行，我会尝试用流来实现

IntStream.range(0, height).parallel().forEach(y -> {    
    IntStream.range(0, width).parallel().forEach(x -> {
        int pixel = sourceArray[sourceOffset ++];              
        destArray[destOffset ++] = (byte) (pixel      );   
        destArray[destOffset ++] = (byte) (pixel >>  8);   
        destArray[destOffset ++] = (byte) (pixel >> 16);   
        destArray[destOffset ++] = (byte) (pixel >> 24);   
    });                                            
    sourceOffset += sourceIncrement;                         
    destOffset += destIncrement;                        
}); 

但这是错误的，因为偏移量会在依赖于它们的代码运行之前增加。是否真的可以使其正确并行运行？

Answer 1

不要执行手动复制。

您可以使用，例如

IntBuffer src = IntBuffer.wrap(sourceArray, sourceOffset, sourceArray.length-sourceOffset);
IntBuffer dst = ByteBuffer.wrap(destArray, destOffset, destArray.length - destOffset)
        .order(ByteOrder.LITTLE_ENDIAN).asIntBuffer();

for(int i = 0; i < height; i++) {
    dst.put(src.limit(src.position()+width));
    src.limit(src.capacity()).position(src.position() + sourceIncrement);
    dst.position(dst.position() + (destIncrement >> 2));
}

这假设 destIncrement 描述的是像素单位，即是四的倍数。此外，它假设数组足够长，以至于有 sourceIncrement 。 destIncrement 最后一行的空间，未写入，但缓冲区不允许设置最后一行之后的位置，否则。

如果数组没有那个空间，则必须在增加位置之前退出最后一行的循环:

if(height > 0) {
    IntBuffer src = IntBuffer.wrap(sourceArray, sourceOffset,
                                                sourceArray.length - sourceOffset);
    IntBuffer dst = ByteBuffer.wrap(destArray, destOffset, destArray.length - destOffset)
            .order(ByteOrder.LITTLE_ENDIAN).asIntBuffer();

    for(int i = 0; ;) {
        dst.put(src.limit(src.position()+width));
        if(++i == height) break;
        src.limit(src.capacity()).position(src.position() + sourceIncrement);
        dst.position(dst.position() + (destIncrement >> 2));
    }
}

尚不清楚这是否会从并行处理中受益，但为了完成，这里有一个并行变体:

IntBuffer src = IntBuffer.wrap(sourceArray, sourceOffset, (width+sourceIncrement)*height);
IntBuffer dst = ByteBuffer.wrap(destArray, destOffset, (width*4+destIncrement)*height)
        .order(ByteOrder.LITTLE_ENDIAN).asIntBuffer();

int srcRow = width + sourceIncrement, dstRow = width + (destIncrement>>2);
IntStream.range(0, height).parallel()
    .forEach(y -> dst.slice().position(y * dstRow)
        .put(src.slice().position(y * srcRow).limit(y * srcRow + width))
    );

由于缓冲区的位置和限制不是线程安全的，因此我们必须创建一个本地缓冲区。该解决方案使用 slice() 封装了 sourceOffset 的初始值。 destOffset，简化转账操作的位置和限制的计算。在传输之前计算它们还可以确保位置永远不会前进到比最后一行的宽度更远的位置。