作者：codestats | 架构设计思想分享者 擅长用最简单的示例复现最核心的架构设计思想
📌 写在前面
Stream API 很多人每天都在用，但能说清楚它“为什么这样设计”的人不多。
今天我们从零手写一个极简版Stream，不依赖任何黑魔法。当你亲手实现过一遍，再看源码会发现——原来就这么回事。
一、先问一个问题：Stream到底解决了什么痛点？
先看一段“朴素”代码：
java
// 需求：找出年龄>18的男性用户，取前10个名字
List<User> users = getUsers();
List<String> result = new ArrayList<>();
int count = 0;
for (User user : users) {
if (user.getAge() > 18 && "男".equals(user.getGender())) {
result.add(user.getName());
count++;
if (count >= 10) break;
}
}
这段代码有什么问题？
| 🔴 代码冗长 | 循环、判断、计数、边界混在一起 |
| 🔴 意图不清 | 需要3秒才能看出“过滤+映射+截断”的逻辑 |
| 🔴 难以扩展 | 想加排序？再加循环？想并行？改到崩溃 |
| 🔴 中间集合 | 每步都产生新List，内存浪费 |
再看Stream版本：
java
List<String> result = users.stream()
.filter(u -> u.getAge() > 18 && "男".equals(u.getGender()))
.map(User::getName)
.limit(10)
.collect(Collectors.toList());
高下立判。但问题来了——它是怎么做到的？
二、核心设计思想（一句话总结）
把“做什么”和“怎么做”分离。你只管声明操作，Stream内部管理迭代、短路、求值时机。
四大关键词
| 🎯 函数式编程 | 把行为（Predicate、Function）当参数传递 |
| 🔗 流水线模式 | 操作不立即执行，而是串成一条链 |
| 😴 惰性求值 | 只有终端操作（forEach/collect）才真正执行 |
| 🔄 内部迭代 | 开发者不用写for循环，Stream替你迭代 |
三、最精简复现：从零实现MiniStream
3.1 定义函数式接口（简化JDK）
java
// 断言：过滤条件
interface MyPredicate<T> { boolean test(T t); }
// 映射：类型转换
interface MyFunction<T, R> { R apply(T t); }
// 消费：最终操作
interface MyConsumer<T> { void accept(T t); }
3.2 核心Stream接口
java
interface MyStream<T> {
// 中间操作（返回Stream，支持链式）
MyStream<T> filter(MyPredicate<T> predicate);
<R> MyStream<R> map(MyFunction<T, R> mapper);
// 终端操作（触发真正计算）
void forEach(MyConsumer<T> action);
}
💡 关键点：filter和map返回的还是MyStream，所以能链式调用。但它们没有真正执行，只是在“记账”。
3.3 最核心的实现类
java
class MyStreamImpl<T> implements MyStream<T> {
private final List<T> source; // 数据源
private final MyPipelineStage<T, T> stage; // 操作链
public static <T> MyStream<T> stream(List<T> list) {
return new MyStreamImpl<>(list, null);
}
private MyStreamImpl(List<T> source, MyPipelineStage<T, T> stage) {
this.source = source;
this.stage = stage;
}
@Override
public MyStream<T> filter(MyPredicate<T> predicate) {
// 记录过滤操作，不执行
MyPipelineStage<T, T> newStage = (value, sink) -> {
if (predicate.test(value)) sink.accept(value);
};
return new MyStreamImpl<>(source, combine(newStage));
}
@Override
public <R> MyStream<R> map(MyFunction<T, R> mapper) {
// 记录映射操作，不执行
MyPipelineStage<R, T> newStage = (value, sink) -> {
R result = mapper.apply(value);
sink.accept(result);
};
return new MyStreamImpl<>(source, combine(newStage));
}
@Override
public void forEach(MyConsumer<T> action) {
// 终端方法：终于执行了！
MyConsumer<T> finalSink = action;
if (stage != null) {
finalSink = value -> stage.process(value, action);
}
for (T item : source) {
finalSink.accept(item);
}
}
// 合并两个阶段（构建责任链）
private MyPipelineStage combine(MyPipelineStage newStage) {
if (stage == null) return newStage;
return (value, sink) -> stage.process(value, v -> newStage.process(v, sink));
}
}
// 流水线阶段接口（核心中的核心）
interface MyPipelineStage<T, R> {
void process(T input, MyConsumer<R> sink);
}
3.4 执行流程图解
text
┌─────────────────────────────────────────────────────────────┐
│ 【构建阶段 – 啥都没执行】 │
└─────────────────────────────────────────────────────────────┘
users.stream() ──→ new MyStreamImpl(source)
│
├─→ filter(…) ──→ 记录 PipelineStage{ if(条件) 传给下游 }
│ (返回新Stream，包装这个Stage)
│
└─→ map(…) ──→ 记录 PipelineStage{ 转换类型，传给下游 }
(再次包装)
📦 此时数据还在source里，一动不动
┌─────────────────────────────────────────────────────────────┐
│ 【触发阶段 – forEach才动真格】 │
└─────────────────────────────────────────────────────────────┘
forEach(action) 被调用
│
├─→ 把最终的action当作sink
│
├─→ 从最外层的stage开始，反向构建调用链：
│ map的stage 调用 → filter的stage调用 → action
│
└─→ for循环遍历source，每个元素走一遍上面的链
具体数据流（以元素 5 为例）：
source中的5
↓
map stage (5 → 25)
↓
filter stage (25>10? ✓)
↓
action (打印 25)
四、为什么能实现惰性求值？
| ❓ 为什么filter/map不执行？ | 它们只是组装了一个函数（PipelineStage），没有调用 |
| ❓ 什么时候执行？ | forEach 触发时，才把函数应用到每个元素上 |
| ❓ 为什么没有中间集合？ | 元素是流式处理：5进去→过滤→映射→打印，一气呵成 |
五、跑起来看看
java
public class Demo {
public static void main(String[] args) {
List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);
MyStreamImpl.stream(numbers)
.filter(n -> n % 2 == 0) // 偶数
.map(n -> n * n) // 平方
.forEach(n -> System.out.print(n + " "));
}
}
// 输出：4 16
执行过程逐行拆解：
| 1 | stream(numbers) | 创建MyStreamImpl，source=[1,2,3,4,5] |
| 2 | filter(偶数) | 记录检查偶数的Stage |
| 3 | map(平方) | 记录平方的Stage，合并成：先过滤再平方 |
| 4 | forEach(打印) | 遍历source，每个元素走链条 |
六、设计思想对照表
| 代码冗长 | 只需声明filter/map/forEach，不写for/if |
| 意图不明确 | .filter().map().forEach() 自解释 |
| 中间集合 | 流式处理，一个元素走完全程才处理下一个 |
| 难以并行 | 替换forEach里的for为多线程分片，使用者无感知 |
七、完整代码（复制可运行）
java
import java.util.*;
// 函数式接口
interface MyPredicate<T> { boolean test(T t); }
interface MyFunction<T, R> { R apply(T t); }
interface MyConsumer<T> { void accept(T t); }
// 流水线阶段
interface MyPipelineStage<T, R> {
void process(T input, MyConsumer<R> sink);
}
// Stream接口
interface MyStream<T> {
MyStream<T> filter(MyPredicate<T> predicate);
<R> MyStream<R> map(MyFunction<T, R> mapper);
void forEach(MyConsumer<T> action);
}
// Stream实现
class MyStreamImpl<T> implements MyStream<T> {
private final List<T> source;
private final MyPipelineStage<T, T> stage;
public static <T> MyStream<T> stream(List<T> list) {
return new MyStreamImpl<>(list, null);
}
private MyStreamImpl(List<T> source, MyPipelineStage<T, T> stage) {
this.source = source;
this.stage = stage;
}
@Override
public MyStream<T> filter(MyPredicate<T> predicate) {
MyPipelineStage<T, T> newStage = (value, sink) -> {
if (predicate.test(value)) sink.accept(value);
};
return new MyStreamImpl<>(source, combine(newStage));
}
@Override
public <R> MyStream<R> map(MyFunction<T, R> mapper) {
MyPipelineStage<R, T> newStage = (value, sink) -> {
R result = mapper.apply(value);
sink.accept(result);
};
return new MyStreamImpl<>(source, combine(newStage));
}
@Override
public void forEach(MyConsumer<T> action) {
MyConsumer<T> finalAction = action;
if (stage != null) {
finalAction = value -> stage.process(value, action);
}
for (T item : source) {
finalAction.accept(item);
}
}
@SuppressWarnings("unchecked")
private <R> MyPipelineStage combine(MyPipelineStage newStage) {
if (stage == null) return newStage;
return (value, sink) -> stage.process((T) value, v -> newStage.process(v, sink));
}
}
// 测试
public class StreamApiCoreDemo {
public static void main(String[] args) {
List<Integer> nums = Arrays.asList(1, 2, 3, 4, 5, 6);
System.out.println("📦 平方后>10的数：");
MyStreamImpl.stream(nums)
.map(n -> n * n)
.filter(n -> n > 10)
.forEach(n -> System.out.print(n + " "));
// 输出：16 25 36
System.out.println("\\n\\n📦 偶数且大于2：");
MyStreamImpl.stream(nums)
.filter(n -> n % 2 == 0)
.filter(n -> n > 2)
.forEach(n -> System.out.print(n + " "));
// 输出：4 6
}
}
📝 最后
Stream API 没有魔法。
它的本质是：你把 filter、map 这些操作组装成一条流水线，但机器没开。直到你按下 forEach 或 collect 这个“启动按钮”，数据才开始从流水线上一个个流过，每个零件处理完立刻传给下一个。
这就是 惰性求值 + 流水线。
善于用最简单的示例复现最核心的思想，从整体上把握它“在干什么”，才能真正理解它。
今天你亲手实现了一个最简版——以后再看Stream源码，心里应该有底了。
点赞 👍 让更多人看到 收藏 ⭐ 方便后续研究 评论 💬 分享你的想法或尝试经验
📌 作者： CodeStats-CSDN博客 擅长用最简单的示例复现最核心的架构设计思想
如果这篇文章对你有帮助，欢迎点赞、收藏、关注。下篇预告：Collector 的设计与 groupingBy 实现。
