下面列出了java.util.function.BinaryOperator#apply ( ) 实例代码，或者点击链接到github查看源代码，也可以在右侧发表评论。
private static BinaryOperator<SymbolDependency> guardedMerge(BinaryOperator<SymbolDependency> original) {
return (a, b) -> {
if (a.getVersion().equals(b.getVersion())) {
return b;
} else {
return original.apply(a, b);
}
};
}
/**
* Constructs a {@code TerminalOp} that implements a functional reduce on
* reference values.
*
* @param <T> the type of the input elements
* @param <U> the type of the result
* @param seed the identity element for the reduction
* @param reducer the accumulating function that incorporates an additional
* input element into the result
* @param combiner the combining function that combines two intermediate
* results
* @return a {@code TerminalOp} implementing the reduction
*/
public static <T, U> TerminalOp<T, U>
makeRef(U seed, BiFunction<U, ? super T, U> reducer, BinaryOperator<U> combiner) {
Objects.requireNonNull(reducer);
Objects.requireNonNull(combiner);
class ReducingSink extends Box<U> implements AccumulatingSink<T, U, ReducingSink> {
@Override
public void begin(long size) {
state = seed;
}
@Override
public void accept(T t) {
state = reducer.apply(state, t);
}
@Override
public void combine(ReducingSink other) {
state = combiner.apply(state, other.state);
}
}
return new ReduceOp<T, U, ReducingSink>(StreamShape.REFERENCE) {
@Override
public ReducingSink makeSink() {
return new ReducingSink();
}
};
}
/**
* Constructs a {@code TerminalOp} that implements a mutable reduce on
* {@code double} values.
*
* @param <R> the type of the result
* @param supplier a factory to produce a new accumulator of the result type
* @param accumulator a function to incorporate an int into an
* accumulator
* @param combiner a function to combine an accumulator into another
* @return a {@code TerminalOp} implementing the reduction
*/
public static <R> TerminalOp<Double, R>
makeDouble(Supplier<R> supplier,
ObjDoubleConsumer<R> accumulator,
BinaryOperator<R> combiner) {
Objects.requireNonNull(supplier);
Objects.requireNonNull(accumulator);
Objects.requireNonNull(combiner);
class ReducingSink extends Box<R>
implements AccumulatingSink<Double, R, ReducingSink>, Sink.OfDouble {
@Override
public void begin(long size) {
state = supplier.get();
}
@Override
public void accept(double t) {
accumulator.accept(state, t);
}
@Override
public void combine(ReducingSink other) {
state = combiner.apply(state, other.state);
}
}
return new ReduceOp<Double, R, ReducingSink>(StreamShape.DOUBLE_VALUE) {
@Override
public ReducingSink makeSink() {
return new ReducingSink();
}
};
}
/**
* Constructs a {@code TerminalOp} that implements a mutable reduce on
* reference values.
*
* @param <T> the type of the input elements
* @param <I> the type of the intermediate reduction result
* @param collector a {@code Collector} defining the reduction
* @return a {@code ReduceOp} implementing the reduction
*/
public static <T, I> TerminalOp<T, I>
makeRef(Collector<? super T, I, ?> collector) {
Supplier<I> supplier = Objects.requireNonNull(collector).supplier();
BiConsumer<I, ? super T> accumulator = collector.accumulator();
BinaryOperator<I> combiner = collector.combiner();
class ReducingSink extends Box<I>
implements AccumulatingSink<T, I, ReducingSink> {
@Override
public void begin(long size) {
state = supplier.get();
}
@Override
public void accept(T t) {
accumulator.accept(state, t);
}
@Override
public void combine(ReducingSink other) {
state = combiner.apply(state, other.state);
}
}
return new ReduceOp<T, I, ReducingSink>(StreamShape.REFERENCE) {
@Override
public ReducingSink makeSink() {
return new ReducingSink();
}
@Override
public int getOpFlags() {
return collector.characteristics().contains(Collector.Characteristics.UNORDERED)
? StreamOpFlag.NOT_ORDERED
: 0;
}
};
}
/**
* Constructs a {@code TerminalOp} that implements a mutable reduce on
* reference values.
*
* @param <T> the type of the input elements
* @param <I> the type of the intermediate reduction result
* @param collector a {@code Collector} defining the reduction
* @return a {@code ReduceOp} implementing the reduction
*/
public static <T, I> TerminalOp<T, I>
makeRef(Collector<? super T, I, ?> collector) {
Supplier<I> supplier = Objects.requireNonNull(collector).supplier();
BiConsumer<I, ? super T> accumulator = collector.accumulator();
BinaryOperator<I> combiner = collector.combiner();
class ReducingSink extends Box<I>
implements AccumulatingSink<T, I, ReducingSink> {
@Override
public void begin(long size) {
state = supplier.get();
}
@Override
public void accept(T t) {
accumulator.accept(state, t);
}
@Override
public void combine(ReducingSink other) {
state = combiner.apply(state, other.state);
}
}
return new ReduceOp<T, I, ReducingSink>(StreamShape.REFERENCE) {
@Override
public ReducingSink makeSink() {
return new ReducingSink();
}
@Override
public int getOpFlags() {
return collector.characteristics().contains(Collector.Characteristics.UNORDERED)
? StreamOpFlag.NOT_ORDERED
: 0;
}
};
}
/**
* Constructs a {@code TerminalOp} that implements a mutable reduce on
* {@code double} values.
*
* @param <R> the type of the result
* @param supplier a factory to produce a new accumulator of the result type
* @param accumulator a function to incorporate an int into an
* accumulator
* @param combiner a function to combine an accumulator into another
* @return a {@code TerminalOp} implementing the reduction
*/
public static <R> TerminalOp<Double, R>
makeDouble(Supplier<R> supplier,
ObjDoubleConsumer<R> accumulator,
BinaryOperator<R> combiner) {
Objects.requireNonNull(supplier);
Objects.requireNonNull(accumulator);
Objects.requireNonNull(combiner);
class ReducingSink extends Box<R>
implements AccumulatingSink<Double, R, ReducingSink>, Sink.OfDouble {
@Override
public void begin(long size) {
state = supplier.get();
}
@Override
public void accept(double t) {
accumulator.accept(state, t);
}
@Override
public void combine(ReducingSink other) {
state = combiner.apply(state, other.state);
}
}
return new ReduceOp<Double, R, ReducingSink>(StreamShape.DOUBLE_VALUE) {
@Override
public ReducingSink makeSink() {
return new ReducingSink();
}
};
}
/**
* Constructs a {@code TerminalOp} that implements a mutable reduce on
* {@code long} values.
*
* @param <R> the type of the result
* @param supplier a factory to produce a new accumulator of the result type
* @param accumulator a function to incorporate an int into an
* accumulator
* @param combiner a function to combine an accumulator into another
* @return a {@code TerminalOp} implementing the reduction
*/
public static <R> TerminalOp<Long, R>
makeLong(Supplier<R> supplier,
ObjLongConsumer<R> accumulator,
BinaryOperator<R> combiner) {
Objects.requireNonNull(supplier);
Objects.requireNonNull(accumulator);
Objects.requireNonNull(combiner);
class ReducingSink extends Box<R>
implements AccumulatingSink<Long, R, ReducingSink>, Sink.OfLong {
@Override
public void begin(long size) {
state = supplier.get();
}
@Override
public void accept(long t) {
accumulator.accept(state, t);
}
@Override
public void combine(ReducingSink other) {
state = combiner.apply(state, other.state);
}
}
return new ReduceOp<Long, R, ReducingSink>(StreamShape.LONG_VALUE) {
@Override
public ReducingSink makeSink() {
return new ReducingSink();
}
};
}
/**
* Constructs a {@code TerminalOp} that implements a functional reduce on
* reference values.
*
* @param <T> the type of the input elements
* @param <U> the type of the result
* @param seed the identity element for the reduction
* @param reducer the accumulating function that incorporates an additional
* input element into the result
* @param combiner the combining function that combines two intermediate
* results
* @return a {@code TerminalOp} implementing the reduction
*/
public static <T, U> TerminalOp<T, U>
makeRef(U seed, BiFunction<U, ? super T, U> reducer, BinaryOperator<U> combiner) {
Objects.requireNonNull(reducer);
Objects.requireNonNull(combiner);
class ReducingSink extends Box<U> implements AccumulatingSink<T, U, ReducingSink> {
@Override
public void begin(long size) {
state = seed;
}
@Override
public void accept(T t) {
state = reducer.apply(state, t);
}
@Override
public void combine(ReducingSink other) {
state = combiner.apply(state, other.state);
}
}
return new ReduceOp<T, U, ReducingSink>(StreamShape.REFERENCE) {
@Override
public ReducingSink makeSink() {
return new ReducingSink();
}
};
}
/**
* Constructs a {@code TerminalOp} that implements a functional reduce on
* reference values producing an optional reference result.
*
* @param <T> The type of the input elements, and the type of the result
* @param operator The reducing function
* @return A {@code TerminalOp} implementing the reduction
*/
public static <T> TerminalOp<T, Optional<T>>
makeRef(BinaryOperator<T> operator) {
Objects.requireNonNull(operator);
class ReducingSink
implements AccumulatingSink<T, Optional<T>, ReducingSink> {
private boolean empty;
private T state;
public void begin(long size) {
empty = true;
state = null;
}
@Override
public void accept(T t) {
if (empty) {
empty = false;
state = t;
} else {
state = operator.apply(state, t);
}
}
@Override
public Optional<T> get() {
return empty ? Optional.empty() : Optional.of(state);
}
@Override
public void combine(ReducingSink other) {
if (!other.empty)
accept(other.state);
}
}
return new ReduceOp<T, Optional<T>, ReducingSink>(StreamShape.REFERENCE) {
@Override
public ReducingSink makeSink() {
return new ReducingSink();
}
};
}
/**
* Constructs a {@code TerminalOp} that implements a functional reduce on
* reference values producing an optional reference result.
*
* @param <T> The type of the input elements, and the type of the result
* @param operator The reducing function
* @return A {@code TerminalOp} implementing the reduction
*/
public static <T> TerminalOp<T, Optional<T>>
makeRef(BinaryOperator<T> operator) {
Objects.requireNonNull(operator);
class ReducingSink
implements AccumulatingSink<T, Optional<T>, ReducingSink> {
private boolean empty;
private T state;
public void begin(long size) {
empty = true;
state = null;
}
@Override
public void accept(T t) {
if (empty) {
empty = false;
state = t;
} else {
state = operator.apply(state, t);
}
}
@Override
public Optional<T> get() {
return empty ? Optional.empty() : Optional.of(state);
}
@Override
public void combine(ReducingSink other) {
if (!other.empty)
accept(other.state);
}
}
return new ReduceOp<T, Optional<T>, ReducingSink>(StreamShape.REFERENCE) {
@Override
public ReducingSink makeSink() {
return new ReducingSink();
}
};
}
/**
* Atomically updates the element at index {@code i} with the
* results of applying the given function to the current and
* given values, returning the updated value. The function should
* be side-effect-free, since it may be re-applied when attempted
* updates fail due to contention among threads. The function is
* applied with the current value at index {@code i} as its first
* argument, and the given update as the second argument.
*
* @param i the index
* @param x the update value
* @param accumulatorFunction a side-effect-free function of two arguments
* @return the updated value
* @since 1.8
*/
public final E accumulateAndGet(int i, E x,
BinaryOperator<E> accumulatorFunction) {
long offset = checkedByteOffset(i);
E prev, next;
do {
prev = getRaw(offset);
next = accumulatorFunction.apply(prev, x);
} while (!compareAndSetRaw(offset, prev, next));
return next;
}
/**
* Atomically updates the field of the given object managed by this
* updater with the results of applying the given function to the
* current and given values, returning the updated value. The
* function should be side-effect-free, since it may be re-applied
* when attempted updates fail due to contention among threads. The
* function is applied with the current value as its first argument,
* and the given update as the second argument.
*
* @param obj An object whose field to get and set
* @param x the update value
* @param accumulatorFunction a side-effect-free function of two arguments
* @return the updated value
* @since 1.8
*/
public final V accumulateAndGet(T obj, V x,
BinaryOperator<V> accumulatorFunction) {
V prev, next;
do {
prev = get(obj);
next = accumulatorFunction.apply(prev, x);
} while (!compareAndSet(obj, prev, next));
return next;
}
/**
* Atomically updates the current value with the results of
* applying the given function to the current and given values,
* returning the updated value. The function should be
* side-effect-free, since it may be re-applied when attempted
* updates fail due to contention among threads. The function
* is applied with the current value as its first argument,
* and the given update as the second argument.
*
* @param x the update value
* @param accumulatorFunction a side-effect-free function of two arguments
* @return the updated value
* @since 1.8
*/
public final V accumulateAndGet(V x,
BinaryOperator<V> accumulatorFunction) {
V prev, next;
do {
prev = get();
next = accumulatorFunction.apply(prev, x);
} while (!compareAndSet(prev, next));
return next;
}
/**
* Returns a {@code Collector} which performs a reduction of its
* input elements under a specified mapping function and
* {@code BinaryOperator}. This is a generalization of
* {@link #reducing(Object, BinaryOperator)} which allows a transformation
* of the elements before reduction.
*
* @apiNote
* The {@code reducing()} collectors are most useful when used in a
* multi-level reduction, downstream of {@code groupingBy} or
* {@code partitioningBy}. To perform a simple map-reduce on a stream,
* use {@link Stream#map(Function)} and {@link Stream#reduce(Object, BinaryOperator)}
* instead.
*
* <p>For example, given a stream of {@code Person}, to calculate the longest
* last name of residents in each city:
* <pre>{@code
* Comparator<String> byLength = Comparator.comparing(String::length);
* Map<City, String> longestLastNameByCity
* = people.stream().collect(groupingBy(Person::getCity,
* reducing(Person::getLastName, BinaryOperator.maxBy(byLength))));
* }</pre>
*
* @param <T> the type of the input elements
* @param <U> the type of the mapped values
* @param identity the identity value for the reduction (also, the value
* that is returned when there are no input elements)
* @param mapper a mapping function to apply to each input value
* @param op a {@code BinaryOperator<U>} used to reduce the mapped values
* @return a {@code Collector} implementing the map-reduce operation
*
* @see #reducing(Object, BinaryOperator)
* @see #reducing(BinaryOperator)
*/
public static <T, U>
Collector<T, ?, U> reducing(U identity,
Function<? super T, ? extends U> mapper,
BinaryOperator<U> op) {
return new CollectorImpl<>(
boxSupplier(identity),
(a, t) -> { a[0] = op.apply(a[0], mapper.apply(t)); },
(a, b) -> { a[0] = op.apply(a[0], b[0]); return a; },
a -> a[0], CH_NOID);
}
/**
* Returns a {@code Collector} which performs a reduction of its
* input elements under a specified {@code BinaryOperator} using the
* provided identity.
*
* @apiNote
* The {@code reducing()} collectors are most useful when used in a
* multi-level reduction, downstream of {@code groupingBy} or
* {@code partitioningBy}. To perform a simple reduction on a stream,
* use {@link Stream#reduce(Object, BinaryOperator)}} instead.
*
* @param <T> element type for the input and output of the reduction
* @param identity the identity value for the reduction (also, the value
* that is returned when there are no input elements)
* @param op a {@code BinaryOperator<T>} used to reduce the input elements
* @return a {@code Collector} which implements the reduction operation
*
* @see #reducing(BinaryOperator)
* @see #reducing(Object, Function, BinaryOperator)
*/
public static <T> Collector<T, ?, T>
reducing(T identity, BinaryOperator<T> op) {
return new CollectorImpl<>(
boxSupplier(identity),
(a, t) -> { a[0] = op.apply(a[0], t); },
(a, b) -> { a[0] = op.apply(a[0], b[0]); return a; },
a -> a[0],
CH_NOID);
}
/**
* Returns a {@code Collector} which performs a reduction of its
* input elements under a specified mapping function and
* {@code BinaryOperator}. This is a generalization of
* {@link #reducing(Object, BinaryOperator)} which allows a transformation
* of the elements before reduction.
*
* @apiNote
* The {@code reducing()} collectors are most useful when used in a
* multi-level reduction, downstream of {@code groupingBy} or
* {@code partitioningBy}. To perform a simple map-reduce on a stream,
* use {@link Stream#map(Function)} and {@link Stream#reduce(Object, BinaryOperator)}
* instead.
*
* <p>For example, given a stream of {@code Person}, to calculate the longest
* last name of residents in each city:
* <pre>{@code
* Comparator<String> byLength = Comparator.comparing(String::length);
* Map<City, String> longestLastNameByCity
* = people.stream().collect(groupingBy(Person::getCity,
* reducing(Person::getLastName, BinaryOperator.maxBy(byLength))));
* }</pre>
*
* @param <T> the type of the input elements
* @param <U> the type of the mapped values
* @param identity the identity value for the reduction (also, the value
* that is returned when there are no input elements)
* @param mapper a mapping function to apply to each input value
* @param op a {@code BinaryOperator<U>} used to reduce the mapped values
* @return a {@code Collector} implementing the map-reduce operation
*
* @see #reducing(Object, BinaryOperator)
* @see #reducing(BinaryOperator)
*/
public static <T, U>
Collector<T, ?, U> reducing(U identity,
Function<? super T, ? extends U> mapper,
BinaryOperator<U> op) {
return new CollectorImpl<>(
boxSupplier(identity),
(a, t) -> { a[0] = op.apply(a[0], mapper.apply(t)); },
(a, b) -> { a[0] = op.apply(a[0], b[0]); return a; },
a -> a[0], CH_NOID);
}
/**
* Atomically updates the field of the given object managed by this
* updater with the results of applying the given function to the
* current and given values, returning the previous value. The
* function should be side-effect-free, since it may be re-applied
* when attempted updates fail due to contention among threads. The
* function is applied with the current value as its first argument,
* and the given update as the second argument.
*
* @param obj An object whose field to get and set
* @param x the update value
* @param accumulatorFunction a side-effect-free function of two arguments
* @return the previous value
* @since 1.8
*/
public final V getAndAccumulate(T obj, V x,
BinaryOperator<V> accumulatorFunction) {
V prev, next;
do {
prev = get(obj);
next = accumulatorFunction.apply(prev, x);
} while (!compareAndSet(obj, prev, next));
return prev;
}
/**
* Atomically updates the element at index {@code i} with the
* results of applying the given function to the current and
* given values, returning the previous value. The function should
* be side-effect-free, since it may be re-applied when attempted
* updates fail due to contention among threads. The function is
* applied with the current value at index {@code i} as its first
* argument, and the given update as the second argument.
*
* @param i the index
* @param x the update value
* @param accumulatorFunction a side-effect-free function of two arguments
* @return the previous value
* @since 1.8
*/
public final E getAndAccumulate(int i, E x,
BinaryOperator<E> accumulatorFunction) {
long offset = checkedByteOffset(i);
E prev, next;
do {
prev = getRaw(offset);
next = accumulatorFunction.apply(prev, x);
} while (!compareAndSetRaw(offset, prev, next));
return prev;
}
/**
* Atomically updates the current value with the results of
* applying the given function to the current and given values,
* returning the updated value. The function should be
* side-effect-free, since it may be re-applied when attempted
* updates fail due to contention among threads. The function
* is applied with the current value as its first argument,
* and the given update as the second argument.
*
* @param x the update value
* @param accumulatorFunction a side-effect-free function of two arguments
* @return the updated value
* @since 1.8
*/
public final V accumulateAndGet(V x,
BinaryOperator<V> accumulatorFunction) {
V prev, next;
do {
prev = get();
next = accumulatorFunction.apply(prev, x);
} while (!compareAndSet(prev, next));
return next;
}
/**
* Atomically updates the current value with the results of
* applying the given function to the current and given values,
* returning the updated value. The function should be
* side-effect-free, since it may be re-applied when attempted
* updates fail due to contention among threads. The function
* is applied with the current value as its first argument,
* and the given update as the second argument.
*
* @param x the update value
* @param accumulatorFunction a side-effect-free function of two arguments
* @return the updated value
* @since 1.8
*/
public final V accumulateAndGet(V x,
BinaryOperator<V> accumulatorFunction) {
V prev, next;
do {
prev = get();
next = accumulatorFunction.apply(prev, x);
} while (!compareAndSet(prev, next));
return next;
}