public static < T > Collector < T , LinkedList < List < T > > , Stream < List < T > > > inBatchesOf ( int itemsPerBatch ) { return Collector . of ( // instantiation LinkedList :: new , // accumulator ( l , i ) -> { if ( l . isEmpty ( ) || l . peekLast ( ) . size ( ) >= itemsPerBatch ) { l . add ( new ArrayList <> ( itemsPerBatch ) ) ; } l . peekLast ( ) . add ( i ) ; } , // combiner ( a , b ) -> { // Merge the two lists so the batches matches the order // of the non-parallel inBatchesOf with (a1..an) + (b1..bn) // as the set of items. It's not extremely efficient, but // works fine as this is not optimized for parallel streams. while ( ! b . isEmpty ( ) ) { for ( T i : b . peekFirst ( ) ) { if ( a . peekLast ( ) . size ( ) >= itemsPerBatch ) { a . add ( new ArrayList <> ( itemsPerBatch ) ) ; } a . peekLast ( ) . add ( i ) ; } b . pollFirst ( ) ; } return a ; } , // finalizer Collection :: stream ) ; }
Collect into batches of max N items per batch . Creates a stream of lists as response .