ZhouXY108/hutool
1
0
Fork 0
mirror of https://gitee.com/chinabugotech/hutool.git synced 2025-04-19 03:01:48 +08:00
Code Issues Packages Projects Releases Wiki Activity

[fix] 处理并行流下zip等依赖于toIdxMap函数时没有保证顺序问题导致的数据顺序错乱的问题

Browse Source
This commit is contained in:
VampireAchao 2023-06-13 11:18:30 +08:00
parent add4a44039
commit 169c625f3b
2 changed files with 29 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
hutool-core/src/main/java/org/dromara/hutool/core/stream/TerminableWrappedStream.java
View File

@ -12,10 +12,10 @@
package org.dromara.hutool.core.stream; package org.dromara.hutool.core.stream;
import org.dromara.hutool.core.array.ArrayUtil;
import org.dromara.hutool.core.lang.Opt; import org.dromara.hutool.core.lang.Opt;
import org.dromara.hutool.core.lang.mutable.MutableInt; import org.dromara.hutool.core.lang.mutable.MutableInt;
import org.dromara.hutool.core.lang.mutable.MutableObj; import org.dromara.hutool.core.lang.mutable.MutableObj;
import org.dromara.hutool.core.array.ArrayUtil;
import java.util.*; import java.util.*;
import java.util.function.*; import java.util.function.*;
@ -214,7 +214,7 @@ public interface TerminableWrappedStream<T, S extends TerminableWrappedStream<T,
*/ */
default <U> Map<Integer, U> toIdxMap(final Function<? super T, ? extends U> valueMapper) { default <U> Map<Integer, U> toIdxMap(final Function<? super T, ? extends U> valueMapper) {
final MutableInt index = new MutableInt(NOT_FOUND_ELEMENT_INDEX); final MutableInt index = new MutableInt(NOT_FOUND_ELEMENT_INDEX);
return EasyStream.of(toList()).toMap(e -> index.incrementAndGet(), valueMapper, (l, r) -> r); return EasyStream.of(parallel().toList()).toMap(e -> index.incrementAndGet(), valueMapper, (l, r) -> r);
} }
// region ============ to zip ============ // region ============ to zip ============
@ -505,8 +505,9 @@ public interface TerminableWrappedStream<T, S extends TerminableWrappedStream<T,
*/ */
default void forEachIdx(final BiConsumer<? super T, Integer> action) { default void forEachIdx(final BiConsumer<? super T, Integer> action) {
Objects.requireNonNull(action); Objects.requireNonNull(action);
if (isParallel()) { final boolean isParallel = isParallel();
EasyStream.of(toIdxMap().entrySet()).parallel(isParallel()) if (isParallel) {
EasyStream.of(toIdxMap().entrySet()).parallel()
.forEach(e -> action.accept(e.getValue(), e.getKey())); .forEach(e -> action.accept(e.getValue(), e.getKey()));
} else { } else {
final MutableInt index = new MutableInt(NOT_FOUND_ELEMENT_INDEX); final MutableInt index = new MutableInt(NOT_FOUND_ELEMENT_INDEX);
@ -522,9 +523,9 @@ public interface TerminableWrappedStream<T, S extends TerminableWrappedStream<T,
*/ */
default void forEachOrderedIdx(final BiConsumer<? super T, Integer> action) { default void forEachOrderedIdx(final BiConsumer<? super T, Integer> action) {
Objects.requireNonNull(action); Objects.requireNonNull(action);
if (isParallel()) { final boolean isParallel = isParallel();
EasyStream.of(toIdxMap().entrySet()) if (isParallel) {
.parallel(isParallel()) EasyStream.of(toIdxMap().entrySet()).parallel()
.forEachOrdered(e -> action.accept(e.getValue(), e.getKey())); .forEachOrdered(e -> action.accept(e.getValue(), e.getKey()));
} else { } else {
final MutableInt index = new MutableInt(NOT_FOUND_ELEMENT_INDEX); final MutableInt index = new MutableInt(NOT_FOUND_ELEMENT_INDEX);

25
hutool-core/src/main/java/org/dromara/hutool/core/stream/TransformableWrappedStream.java
View File

@ -12,6 +12,7 @@
package org.dromara.hutool.core.stream; package org.dromara.hutool.core.stream;
import org.dromara.hutool.core.array.ArrayUtil;
import org.dromara.hutool.core.collection.ListUtil; import org.dromara.hutool.core.collection.ListUtil;
import org.dromara.hutool.core.collection.iter.IterUtil; import org.dromara.hutool.core.collection.iter.IterUtil;
import org.dromara.hutool.core.lang.Console; import org.dromara.hutool.core.lang.Console;
@ -19,7 +20,6 @@ import org.dromara.hutool.core.lang.mutable.MutableInt;
import org.dromara.hutool.core.lang.mutable.MutableObj; import org.dromara.hutool.core.lang.mutable.MutableObj;
import org.dromara.hutool.core.map.MapUtil; import org.dromara.hutool.core.map.MapUtil;
import org.dromara.hutool.core.map.SafeConcurrentHashMap; import org.dromara.hutool.core.map.SafeConcurrentHashMap;
import org.dromara.hutool.core.array.ArrayUtil;
import java.util.*; import java.util.*;
import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicBoolean;
@ -279,10 +279,11 @@ public interface TransformableWrappedStream<T, S extends TransformableWrappedStr
*/ */
default S peekIdx(final BiConsumer<? super T, Integer> action) { default S peekIdx(final BiConsumer<? super T, Integer> action) {
Objects.requireNonNull(action); Objects.requireNonNull(action);
if (isParallel()) { final boolean isParallel = isParallel();
if (isParallel) {
final Map<Integer, T> idxMap = easyStream().toIdxMap(); final Map<Integer, T> idxMap = easyStream().toIdxMap();
return wrap(EasyStream.of(idxMap.entrySet()) return wrap(EasyStream.of(idxMap.entrySet())
.parallel(isParallel()) .parallel()
.peek(e -> action.accept(e.getValue(), e.getKey())) .peek(e -> action.accept(e.getValue(), e.getKey()))
.map(Map.Entry::getValue)); .map(Map.Entry::getValue));
} else { } else {
@ -383,10 +384,10 @@ public interface TransformableWrappedStream<T, S extends TransformableWrappedStr
*/ */
default S filterIdx(final BiPredicate<? super T, Integer> predicate) { default S filterIdx(final BiPredicate<? super T, Integer> predicate) {
Objects.requireNonNull(predicate); Objects.requireNonNull(predicate);
if (isParallel()) { final boolean isParallel = isParallel();
if (isParallel) {
final Map<Integer, T> idxMap = easyStream().toIdxMap(); final Map<Integer, T> idxMap = easyStream().toIdxMap();
return wrap(EasyStream.of(idxMap.entrySet()) return wrap(EasyStream.of(idxMap.entrySet()).parallel()
.parallel(isParallel())
.filter(e -> predicate.test(e.getValue(), e.getKey())) .filter(e -> predicate.test(e.getValue(), e.getKey()))
.map(Map.Entry::getValue)); .map(Map.Entry::getValue));
} else { } else {
@ -440,10 +441,10 @@ public interface TransformableWrappedStream<T, S extends TransformableWrappedStr
*/ */
default <R> EasyStream<R> flatMapIdx(final BiFunction<? super T, Integer, ? extends Stream<? extends R>> mapper) { default <R> EasyStream<R> flatMapIdx(final BiFunction<? super T, Integer, ? extends Stream<? extends R>> mapper) {
Objects.requireNonNull(mapper); Objects.requireNonNull(mapper);
if (isParallel()) { final boolean isParallel = isParallel();
if (isParallel) {
final Map<Integer, T> idxMap = easyStream().toIdxMap(); final Map<Integer, T> idxMap = easyStream().toIdxMap();
return EasyStream.of(idxMap.entrySet()) return EasyStream.of(idxMap.entrySet()).parallel()
.parallel(isParallel())
.flatMap(e -> mapper.apply(e.getValue(), e.getKey())); .flatMap(e -> mapper.apply(e.getValue(), e.getKey()));
} else { } else {
final MutableInt index = new MutableInt(NOT_FOUND_ELEMENT_INDEX); final MutableInt index = new MutableInt(NOT_FOUND_ELEMENT_INDEX);
@ -553,10 +554,10 @@ public interface TransformableWrappedStream<T, S extends TransformableWrappedStr
*/ */
default <R> EasyStream<R> mapIdx(final BiFunction<? super T, Integer, ? extends R> mapper) { default <R> EasyStream<R> mapIdx(final BiFunction<? super T, Integer, ? extends R> mapper) {
Objects.requireNonNull(mapper); Objects.requireNonNull(mapper);
if (isParallel()) { final boolean isParallel = isParallel();
if (isParallel) {
final Map<Integer, T> idxMap = easyStream().toIdxMap(); final Map<Integer, T> idxMap = easyStream().toIdxMap();
return EasyStream.of(idxMap.entrySet()) return EasyStream.of(idxMap.entrySet()).parallel()
.parallel(isParallel())
.map(e -> mapper.apply(e.getValue(), e.getKey())); .map(e -> mapper.apply(e.getValue(), e.getKey()));
} else { } else {
final MutableInt index = new MutableInt(NOT_FOUND_ELEMENT_INDEX); final MutableInt index = new MutableInt(NOT_FOUND_ELEMENT_INDEX);