17370845950

1. keyMapper (键映射器): 一个 Function，用于从输入元素 T 中提取 Map 的键 K。
2. valueMapper (值映射器): 一个 Function，用于从输入元素 T 中提取 Map 的值 U。
3. mergeFunction (合并函数): 一个 BinaryOperator，当遇到重复键时，它定义了如何合并旧值和新值。这是实现值累加的关键。
4. mapFactory (Map 工厂): 一个 Supplier，用于提供一个新的 Map 实例。这是本文要重点讨论的部分，它决定了 Map 的具体实现类型以及是否避免外部状态。

正确处理重复键值累加

假设我们有一个 Position 对象的列表，每个 Position 包含 assetId、currencyId 和 value。我们的目标是创建一个 Map，其中 PositionKey 由 assetId 和 currencyId 组成，如果存在相同的 PositionKey，则将其 value 进行累加。

初始尝试与改进空间

在不熟悉 mapFactory 参数时，开发者可能会尝试在 Stream 外部创建一个 Map，然后将其传递给 toMap 的 mapFactory，例如：

public Map getMap(final Long portfolioId) {
    final Map map = new HashMap<>(); // 外部创建 Map

    return getPositions(portfolioId).stream()
        .collect(
            Collectors.toMap(
                position -> new PositionKey(position.getAssetId(), position.getCurrencyId()),
                position -> position.getValue(),
                (oldValue, newValue) -> oldValue.add(newValue),
                () -> map // 将外部 Map 传递给 mapFactory
            ));
}

这种做法虽然在某些情况下可以工作，但它违背了 Stream API 的函数式编程理念，将 Stream 操作与外部的可变状态紧密耦合。更重要的是，在并行 Stream 处理中，这种方式可能导致不可预测的行为或线程安全问题。

推荐的解决方案：使用 HashMap::new

正确的做法是让 mapFactory 提供一个全新的 Map 实例，而不是引用外部已存在的 Map。这可以通过方法引用 HashMap::new 或 Lambda 表达式 () -> new HashMap() 来实现。这样，Stream 内部会负责创建和管理 Map，保持了操作的纯粹性和独立性。

以下是优化后的代码示例：

import java.math.BigDecimal;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * 模拟 PositionKey 类，作为 Map 的键
 */
class PositionKey {
    String assetId;
    String currencyId;

    public PositionKey(String assetId, String currencyId) {
        this.assetId = assetId;
        this.currencyId = currencyId;
    }

    // 必须重写 equals 和 hashCode，因为 PositionKey 将作为 Map 的键
    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        PositionKey that = (PositionKey) o;
        return Objects.equals(assetId, that.assetId) &&
               Objects.equals(currencyId, that.currencyId);
    }

    @Override
    public int hashCode() {
        return Objects.hash(assetId, currencyId);
    }

    @Override
    public String toString() {
        return "PositionKey{" +
               "assetId='" + assetId + '\'' +
               ", currencyId='" + currencyId + '\'' +
               '}';
    }
}

/**
 * 模拟 Position 类，包含资产信息和值
 */
class Position {
    String assetId;
    String currencyId;
    BigDecimal value; // 使用 BigDecimal 处理金额，避免浮点数精度问题

    public Position(String assetId, String currencyId, BigDecimal value) {
        this.assetId = assetId;
        this.currencyId = currencyId;
    }

    public String getAssetId() { return assetId; }
    public String getCurrencyId() { return currencyId; }
    public BigDecimal getValue() { return value; }
}

public class StreamAggregationTutorial {

    // 模拟获取头寸列表的方法
    private List getPositions(Long portfolioId) {
        // 实际应用中这里会从数据库或其他数据源获取数据
        // 示例数据，包含重复的 PositionKey
        return List.of(
            new Position("AAPL", "USD", new BigDecimal("100.50")),
            new Position("GOOG", "USD", new BigDecimal("200.75")),
            new Position("AAPL", "USD", new BigDecimal("50.25")), // 键重复，值需要累加
            new Position("TSLA", "EUR", new BigDecimal("75.00")),
            new Position("GOOG", "USD", new BigDecimal("10.00"))  // 键重复，值需要累加
        );
    }

    /**
     * 使用 Stream API 聚合头寸数据到 Map，并累加重复键的值。
     *
     * @param portfolioId 投资组合ID
     * @return 聚合后的 Map
     */
    public Map getAggregatedPositionsMap(final Long portfolioId) {
        return getPositions(portfolioId).stream()
            .collect(
                Collectors.toMap(
                    position -> new PositionKey(position.getAssetId(), position.getCurrencyId()), // keyMapper: 创建 PositionKey 作为键
                    Position::getValue, // valueMapper: 提取 Position 的值
                    (oldValue, newValue) -> oldValue.add(newValue), // mergeFunction: 当键重复时，将旧值与新值相加
                    HashMap::new // mapFactory: 提供一个新的 HashMap 实例
                )
            );
    }

    public static void main(String[] args) {
        StreamAggregationTutorial example = new StreamAggregationTutorial();
        Map aggregatedMap = example.getAggregatedPositionsMap(123L);

        System.out.println("聚合后的头寸映射：");
        aggregatedMap.forEach((key, value) -> System.out.println(key + " -> " + value));

        // 预期输出示例:
        // PositionKey{assetId='AAPL', currencyId='USD'} -> 150.75
        // PositionKey{assetId='GOOG', currencyId='USD'} -> 210.75
        // PositionKey{assetId='TSLA', currencyId='EUR'} -> 75.00
    }
}

注意事项与最佳实践

1. PositionKey 的 equals() 和 hashCode()： 当自定义对象作为 Map 的键时，务必正确重写 equals() 和 hashCode() 方法。这是 Map 正确识别键、避免存储重复键的关键。在上述示例中，PositionKey 类已经包含了这些方法的实现。
2. BigDecimal 的使用： 对于涉及金额计算的场景，强烈推荐使用 BigDecimal 而不是 double 或 float，以避免浮点数精度问题。BigDecimal 的 add() 方法返回一个新的 BigDecimal 实例，因为它本身是不可变的。
3. mergeFunction 的健壮性： Collectors.toMap 的 mergeFunction 在被调用时，oldValue 参数保证不会是 null，因为它只在键已经存在于 Map 中时才会被调用。因此，oldValue != null 的检查是多余的。
4. mapFactory 的选择： HashMap::new 是最常见的选择，因为它提供了 O(1) 的平均时间复杂度。如果需要保持键的插入顺序，可以使用 LinkedHashMap::new；如果需要键的自然排序或自定义排序，可以使用 TreeMap::new。
5. 并行 Stream 与 mapFactory： 当使用并行 Stream (parallelStream()) 时，Collectors.toMap 会为每个并行处理分支创建独立的 Map 实例，并在最后将这些 Map 合并。mapFactory 提供的就是这些独立 Map 的创建方式。因此，提供一个能够创建新、空 Map 实例的 Supplier 是至关重要的，以确保线程安全和正确性。

总结

通过正确使用 Collectors.toMap 的 keyMapper、valueMapper、mergeFunction 和 mapFactory 参数，我们可以以一种声明式、高效且函数式的方式，将数据流聚合到 Map 中，并优雅地处理重复键的值累加问题。特别是将 mapFactory 设置为 HashMap::new (或任何其他 Map 实现的构造器引用)，能够确保 Stream 操作的独立性，避免不必要的外部状态依赖，并为并行处理奠定良好基础。掌握这些技巧，将使您的 Java Stream 代码更加简洁、健壮和易于维护。