Stream API 高级详解

// 中间操作不会执行，直到终止操作触发
Stream<String> stream = names.stream()
    .filter(n -> {
        System.out.println("filter: " + n);  // 此时不打印
        return n.length() > 2;
    })
    .map(n -> {
        System.out.println("map: " + n);     // 此时不打印
        return n.toUpperCase();
    });
// 到此为止，没有一行打印输出！

// 终止操作触发整个 pipeline 执行
List<String> result = stream.collect(Collectors.toList());
// 现在才依次输出 filter 和 map

// limit(2) 短路：只处理前两个元素就停止
Stream.of(1, 2, 3, 4, 5)
    .peek(x -> System.out.println("处理: " + x))
    .limit(2)
    .collect(Collectors.toList());
// 输出：
// 处理: 1
// 处理: 2
// (3, 4, 5 不会被处理！)

// parallelStream 使用 ForkJoinPool.commonPool()
// 适合 CPU 密集型任务，不适合 I/O 密集型
// 注意线程安全问题：避免在并行流中使用非线程安全集合
// 错误示例：
List<String> list = new ArrayList<>();  // 非线程安全！
stream.parallel().forEach(list::add);   // 可能导致数据丢失或异常

// 正确示例：
List<String> list = stream.parallel().collect(Collectors.toList());  // 线程安全

import java.util.*;
import java.util.stream.*;
import java.util.function.*;

/**
 * 飞翔科技 2024 Q1 员工绩效数据分析
 * 演示 Stream API 各种中间操作和终止操作
 */
class FeixiangEmployee {
    private int id;
    private String name;
    private String department;   // 研发部、市场部、财务部、人事部
    private String city;         // 北京、上海、深圳、杭州
    private double salary;
    private int age;
    private String level;        // P4, P5, P6, P7, P8
    private double performanceScore; // 绩效评分 0-100

    public FeixiangEmployee(int id, String name, String department, String city,
                            double salary, int age, String level, double performanceScore) {
        this.id = id; this.name = name; this.department = department;
        this.city = city; this.salary = salary; this.age = age;
        this.level = level; this.performanceScore = performanceScore;
    }

    // Getters
    public int getId() { return id; }
    public String getName() { return name; }
    public String getDepartment() { return department; }
    public String getCity() { return city; }
    public double getSalary() { return salary; }
    public int getAge() { return age; }
    public String getLevel() { return level; }
    public double getPerformanceScore() { return performanceScore; }

    @Override
    public String toString() {
        return String.format("%s[%s|%s|¥%.2f|%d岁|%s|%.1f分]",
                name, department, city, salary, age, level, performanceScore);
    }
}

public class FeixiangStreamDemo {
    public static void main(String[] args) {
        System.out.println("========== 飞翔科技 Q1 员工数据分析 ==========\n");

        // 构造测试数据
        List<FeixiangEmployee> employees = Arrays.asList(
            new FeixiangEmployee(1001, "大翔",   "研发部", "北京", 18888.88, 30, "P7", 92.5),
            new FeixiangEmployee(1002, "白歌",   "研发部", "深圳", 15888.88, 28, "P6", 88.0),
            new FeixiangEmployee(1003, "小崔",   "研发部", "北京", 8888.88,  24, "P4", 75.5),
            new FeixiangEmployee(1004, "朱璐",   "研发部", "杭州", 12888.88, 26, "P5", 85.0),
            new FeixiangEmployee(1005, "Frank",  "市场部", "上海", 28888.88, 38, "P8", 95.0),
            new FeixiangEmployee(1006, "黄俪",   "人事部", "北京", 10888.88, 32, "P6", 78.5),
            new FeixiangEmployee(1007, "李眉",   "财务部", "上海", 13888.88, 29, "P6", 90.0),
            new FeixiangEmployee(1008, "孔蓝",   "市场部", "深圳", 16888.88, 31, "P6", 82.0),
            new FeixiangEmployee(1009, "刘洋",   "研发部", "北京", 9888.88,  25, "P5", 70.0),
            new FeixiangEmployee(1010, "张伟",   "财务部", "杭州", 11888.88, 27, "P5", 86.5)
        );

        // ===== 1. 基础过滤与映射 =====
        System.out.println("【1> 研发部薪资高于10000的员工（按薪资降序）】");
        employees.stream()
            .filter(e -> "研发部".equals(e.getDepartment()))
            .filter(e -> e.getSalary() > 10000)
            .sorted(Comparator.comparingDouble(FeixiangEmployee::getSalary).reversed())
            .forEach(e -> System.out.printf("  %s - ¥%.2f\n", e.getName(), e.getSalary()));

        // ===== 2. map + distinct =====
        System.out.println("\n【2> 所有部门列表（去重排序）】");
        String departments = employees.stream()
            .map(FeixiangEmployee::getDepartment)
            .distinct()
            .sorted()
            .collect(Collectors.joining(" | "));
        System.out.println("  " + departments);

        // ===== 3. groupingBy 单级分组 =====
        System.out.println("\n【3> 按部门分组】");
        Map<String, List&lt;FeixiangEmployee&gt;> byDept = employees.stream()
            .collect(Collectors.groupingBy(FeixiangEmployee::getDepartment));
        byDept.forEach((dept, emps) -> {
            String names = emps.stream().map(FeixiangEmployee::getName)
                    .collect(Collectors.joining(", "));
            System.out.printf("  %s: [%s] (共 %d 人)\n", dept, names, emps.size());
        });

        // ===== 4. groupingBy 多级分组（部门 + 城市） =====
        System.out.println("\n【4> 按部门再按城市二级分组】");
        Map<String, Map<String, List&lt;FeixiangEmployee&gt;>> byDeptCity = employees.stream()
            .collect(Collectors.groupingBy(
                FeixiangEmployee::getDepartment,
                Collectors.groupingBy(FeixiangEmployee::getCity)
            ));
        byDeptCity.forEach((dept, cityMap) -> {
            System.out.println("  " + dept + ":");
            cityMap.forEach((city, emps) -> {
                System.out.printf("    ├─ %s: %s\n", city,
                    emps.stream().map(FeixiangEmployee::getName)
                        .collect(Collectors.joining(", ")));
            });
        });

        // ===== 5. 按部门统计薪资汇总 =====
        System.out.println("\n【5> 各部门薪资统计汇总】");
        Map&lt;String, DoubleSummaryStatistics&gt; salaryStats = employees.stream()
            .collect(Collectors.groupingBy(
                FeixiangEmployee::getDepartment,
                Collectors.summarizingDouble(FeixiangEmployee::getSalary)
            ));
        System.out.printf("  %-8s %8s %10s %10s %10s %10s\n",
                "部门", "人数", "总薪资", "平均薪资", "最高", "最低");
        System.out.println("  " + "-".repeat(58));
        salaryStats.forEach((dept, stats) -> {
            System.out.printf("  %-8s %8d %10.2f %10.2f %10.2f %10.2f\n",
                dept, stats.getCount(),
                stats.getSum(), stats.getAverage(),
                stats.getMax(), stats.getMin());
        });

        // ===== 6. partitioningBy 分区 =====
        System.out.println("\n【6> 按薪资是否超过15000分区】");
        Map<Boolean, List&lt;FeixiangEmployee&gt;> partitioned = employees.stream()
            .collect(Collectors.partitioningBy(e -> e.getSalary() > 15000));
        System.out.printf("  高薪资 (>¥15000): %s\n",
            partitioned.get(true).stream()
                .map(e -> e.getName() + " ¥" + e.getSalary())
                .collect(Collectors.joining(", ")));
        System.out.printf("  普通薪资: %s\n",
            partitioned.get(false).stream()
                .map(e -> e.getName() + " ¥" + e.getSalary())
                .collect(Collectors.joining(", ")));

        // ===== 7. reduce 归约 =====
        System.out.println("\n【7> 公司总薪资池与平均薪资】");
        double totalSalary = employees.stream()
            .map(FeixiangEmployee::getSalary)
            .reduce(0.0, Double::sum);
        double avgSalary = totalSalary / employees.size();
        System.out.printf("  总薪资池: ¥%.2f | 公司平均: ¥%.2f\n", totalSalary, avgSalary);

        // ===== 8. flatMap 示例 =====
        System.out.println("\n【8> flatMap: 每个部门的技能关键词扁平化】");
        Map<String, List&lt;String&gt;> deptSkills = new LinkedHashMap<>();
        deptSkills.put("研发部", Arrays.asList("Java", "Spring", "MySQL", "Docker"));
        deptSkills.put("市场部", Arrays.asList("SEO", "SEM", "数据分析"));
        deptSkills.put("财务部", Arrays.asList("Excel", "SAP", "税务"));
        deptSkills.put("人事部", Arrays.asList("招聘", "培训", "绩效"));

        // 获取所有部门的所有技能（去重）
        List&lt;String&gt; allSkills = deptSkills.values().stream()
            .flatMap(Collection::stream)
            .distinct()
            .sorted()
            .collect(Collectors.toList());
        System.out.println("  全公司技能池: " + allSkills);

        // ===== 9. peek 调试 =====
        System.out.println("\n【9> peek 调试流处理过程】");
        long highPerfCount = employees.stream()
            .filter(e -> e.getPerformanceScore() > 85)
            .peek(e -> System.out.printf("  [peek] 高绩效: %s (%.1f分)\n",
                    e.getName(), e.getPerformanceScore()))
            .filter(e -> "研发部".equals(e.getDepartment()))
            .peek(e -> System.out.printf("  [peek] 研发部高绩效: %s\n", e.getName()))
            .count();
        System.out.println("  研发部高绩效人数: " + highPerfCount);

        // ===== 10. Collectors.mapping + groupingBy 组合 =====
        System.out.println("\n【10> 各部门绩效最高员工姓名】");
        Map&lt;String, String&gt; topPerformerByDept = employees.stream()
            .collect(Collectors.groupingBy(
                FeixiangEmployee::getDepartment,
                Collectors.collectingAndThen(
                    Collectors.maxBy(
                        Comparator.comparingDouble(FeixiangEmployee::getPerformanceScore)
                    ),
                    opt -> opt.map(FeixiangEmployee::getName).orElse("无")
                )
            ));
        topPerformerByDept.forEach((dept, name) ->
            System.out.printf("  %s: %s\n", dept, name));

        // ===== 11. IntStream 基本类型流 =====
        System.out.println("\n【11> IntStream 年龄统计分析】");
        IntSummaryStatistics ageStats = employees.stream()
            .mapToInt(FeixiangEmployee::getAge)
            .summaryStatistics();
        System.out.printf("  年龄范围: %d ~ %d | 平均: %.1f岁 | 总和: %d\n",
                ageStats.getMin(), ageStats.getMax(),
                ageStats.getAverage(), ageStats.getSum());

        System.out.println("\n========== End ==========");
    }
}

import java.util.*;
import java.util.stream.*;

/**
 * 飞翔科技商城订单多维分析 —— 实战 Stream API
 */
class Order {
    private String orderId;
    private String customerName;
    private String productCategory; // 电子产品、图书、服装、食品
    private String productName;
    private int quantity;
    private double unitPrice;
    private String month; // 2024-01, 2024-02, 2024-03
    private String paymentMethod; // 微信、支付宝、银行卡

    public Order(String orderId, String customerName, String productCategory,
                 String productName, int quantity, double unitPrice,
                 String month, String paymentMethod) {
        this.orderId = orderId; this.customerName = customerName;
        this.productCategory = productCategory; this.productName = productName;
        this.quantity = quantity; this.unitPrice = unitPrice;
        this.month = month; this.paymentMethod = paymentMethod;
    }

    public double getTotalAmount() { return quantity * unitPrice; }
    public String getProductCategory() { return productCategory; }
    public String getProductName() { return productName; }
    public int getQuantity() { return quantity; }
    public double getUnitPrice() { return unitPrice; }
    public String getMonth() { return month; }
    public String getPaymentMethod() { return paymentMethod; }
    public String getCustomerName() { return customerName; }
}

public class OrderAnalysisDemo {
    public static void main(String[] args) {
        System.out.println("========== 飞翔科技商城订单分析 ==========\n");

        List<Order> orders = Arrays.asList(
            new Order("ORD001", "大翔", "电子产品", "机械键盘",   1, 188.88, "2024-01", "微信"),
            new Order("ORD002", "白歌", "电子产品", "显示器",     1, 1288.88, "2024-01", "支付宝"),
            new Order("ORD003", "小崔", "图书",    "Java编程思想", 2, 88.88, "2024-01", "微信"),
            new Order("ORD004", "朱璐", "服装",    "T恤",         3, 68.88, "2024-01", "银行卡"),
            new Order("ORD005", "大翔", "食品",    "坚果礼盒",   2, 128.88, "2024-02", "微信"),
            new Order("ORD006", "Frank","电子产品", "平板电脑",   1, 3888.88, "2024-02", "支付宝"),
            new Order("ORD007", "黄俪", "服装",    "衬衫",       2, 158.88, "2024-02", "微信"),
            new Order("ORD008", "李眉", "图书",    "Effective Java", 1, 78.88, "2024-02", "支付宝"),
            new Order("ORD009", "孔蓝", "电子产品", "蓝牙耳机",   1, 288.88, "2024-03", "银行卡"),
            new Order("ORD010", "大翔", "食品",    "咖啡豆",     3, 58.88, "2024-03", "微信"),
            new Order("ORD011", "白歌", "图书",    "Clean Code", 2, 68.88, "2024-03", "支付宝"),
            new Order("ORD012", "朱璐", "服装",    "运动鞋",     1, 388.88, "2024-03", "微信")
        );

        // 1. 各品类销售额排行
        System.out.println("【1> 各品类销售额排行】");
        Map<String, Double> categorySales = orders.stream()
            .collect(Collectors.groupingBy(
                Order::getProductCategory,
                Collectors.summingDouble(Order::getTotalAmount)
            ));
        categorySales.entrySet().stream()
            .sorted(Map.Entry<String, Double>comparingByValue().reversed())
            .forEach(e -> System.out.printf("  %s: ¥%.2f\n", e.getKey(), e.getValue()));

        // 2. 月度销售额趋势
        System.out.println("\n【2> 各月度销售额趋势】");
        Map<String, Double> monthlySales = orders.stream()
            .collect(Collectors.groupingBy(
                Order::getMonth,
                TreeMap::new,  // 按月份排序
                Collectors.summingDouble(Order::getTotalAmount)
            ));
        monthlySales.forEach((month, sales) ->
            System.out.printf("  %s: ¥%.2f\n", month, sales));

        // 3. 支付方式占比
        System.out.println("\n【3> 支付方式分布】");
        double totalAmount = orders.stream()
                .mapToDouble(Order::getTotalAmount).sum();
        Map<String, Double> paymentDistribution = orders.stream()
            .collect(Collectors.groupingBy(
                Order::getPaymentMethod,
                Collectors.summingDouble(Order::getTotalAmount)
            ));
        paymentDistribution.forEach((method, amount) ->
            System.out.printf("  %s: ¥%.2f (%.1f%%)\n",
                method, amount, amount / totalAmount * 100));

        // 4. 客户消费排行（Top 3）
        System.out.println("\n【4> 客户消费排行 Top 3】");
        Map<String, Double> customerSpending = orders.stream()
            .collect(Collectors.groupingBy(
                Order::getCustomerName,
                Collectors.summingDouble(Order::getTotalAmount)
            ));
        customerSpending.entrySet().stream()
            .sorted(Map.Entry<String, Double>comparingByValue().reversed())
            .limit(3)
            .forEach(e -> System.out.printf("  %s: ¥%.2f\n", e.getKey(), e.getValue()));

        // 5. 每个品类下价格最高的商品
        System.out.println("\n【5> 每个品类下最贵的商品】");
        Map<String, String> mostExpensive = orders.stream()
            .collect(Collectors.groupingBy(
                Order::getProductCategory,
                Collectors.collectingAndThen(
                    Collectors.maxBy(Comparator.comparingDouble(Order::getUnitPrice)),
                    opt -> opt.map(o -> o.getProductName() + " ¥" + o.getUnitPrice()).orElse("无")
                )
            ));
        mostExpensive.forEach((cat, prod) ->
            System.out.printf("  %s: %s\n", cat, prod));

        // 6. 订单金额分段统计
        System.out.println("\n【6> 订单金额分段（<200 / 200~500 / >500）】");
        Map&lt;String, Long&gt; amountSegments = orders.stream()
            .collect(Collectors.groupingBy(order -> {
                double amt = order.getTotalAmount();
                if (amt < 200) return "小额 (<200)";
                else if (amt <= 500) return "中额 (200~500)";
                else return "大额 (>500)";
            }, Collectors.counting()));
        amountSegments.forEach((seg, cnt) ->
            System.out.printf("  %s: %d 笔\n", seg, cnt));

        // 7. 是否存在单笔超过3000的订单
        System.out.println("\n【7> 是否存在大额订单（>¥3000）？】");
        boolean hasLarge = orders.stream()
                .anyMatch(o -> o.getTotalAmount() > 3000);
        Optional&lt;Order&gt; largeOrder = orders.stream()
                .filter(o -> o.getTotalAmount() > 3000)
                .findFirst();
        largeOrder.ifPresent(o ->
            System.out.printf("  是！%s 购买了 %s，金额 ¥%.2f\n",
                o.getCustomerName(), o.getProductName(), o.getTotalAmount()));

        System.out.println("\n========== 飞翔科技 Q1 数据分析报告完毕 ==========");
    }
}

// 错误示范
Stream<String> stream = list.stream();
stream.forEach(System.out::println);  // 第一次消费 OK
stream.forEach(System.out::println);  // ← IllegalStateException: stream has already been operated upon or closed

// 正确做法：重新获取 Stream
list.stream().forEach(System.out::println);
list.stream().forEach(System.out::println);
// 或者将中间操作链赋值给 Supplier
Supplier<Stream&lt;String&gt;> supplier = () -> list.stream().filter(s -> s.length() > 2);
supplier.get().forEach(System.out::println); // OK
supplier.get().count();                       // OK

// 如果 key 重复，toMap 默认抛异常
List<Order> orders = Arrays.asList(
    new Order("ORD001", "大翔", "电子产品", "键盘", 1, 188.88, "2024-01", "微信"),
    new Order("ORD001", "大翔", "图书", "Java书", 1, 88.88, "2024-01", "微信")
);

// 错误：重复 key "大翔" → IllegalStateException
// Map<String, Order> map = orders.stream()
//     .collect(Collectors.toMap(Order::getCustomerName, Function.identity()));

// 正确：提供 merge function
Map<String, Order> map = orders.stream()
    .collect(Collectors.toMap(
        Order::getCustomerName,
        Function.identity(),
        (existing, replacement) -> replacement  // 保留后者
    ));

// 错误示范
List<Integer> list = new ArrayList<>();  // 非线程安全
IntStream.range(0, 10000)
    .parallel()
    .forEach(list::add);  // ← 数据丢失/ArrayIndexOutOfBoundsException
System.out.println(list.size());  // 可能小于 10000

// 正确做法
List<Integer> list = IntStream.range(0, 10000)
    .parallel()
    .boxed()
    .collect(Collectors.toList());  // 线程安全

// sorted 是全量排序，即使是 sorted(...).limit(5)，也会对整个流排序
// 对于大数据集的 Top-N 需求，应该使用 PriorityQueue 而非 Stream
// 或者使用 Google Guava 的 Ordering.greatestOf()

// map：每个员工→姓名
employees.stream().map(Emp::getName).collect(toList());
// ["大翔", "白歌", ...]

// flatMap：每个员工→技能列表→合并为一个技能流
employees.stream().flatMap(e -> e.getSkills().stream()).distinct().collect(toList());
// ["Java", "Spring", "MySQL", ...]

Map<String, Map<String, List<Emp>>>> = employees.stream()
    .collect(Collectors.groupingBy(
        Emp::getDept,              // 第一级：部门
        Collectors.groupingBy(      // 第二级：城市
            Emp::getCity
        )
    ));

// 对应 SQL: SELECT dept, AVG(salary) FROM emp GROUP BY dept
Map<String, Double> = employees.stream()
    .collect(Collectors.groupingBy(
        Emp::getDept,
        Collectors.averagingDouble(Emp::getSalary)
    ));