请问如何自己实现Java8的Stream？

就是要实现基本的of(),

concat(),

map(),

filter()等方法，实在没有思路，求执教

我写了一份简单代码供参考~

不过如果要优化性能的话，实现上会复杂很多：
(1) 需要特化（specialize）基础类型（例如int, double）的实现，参照Java的IntStream、DoubleStream
(2) 需要考虑lazy evaluation，以减少物化（materialize）数据造成的对内存带宽的冲击和对cache locality的破坏
(3) 如果要做lazy evaluation，实际上就是要生成中间的query plan，那么可以进行进一步的query optimization，比如filter pushdown
(4) 需要考虑对多核并行执行的支持
(5) 再加上对分布式的支持，你就得到了类似Spark的东西..

1. import java.util.ArrayList;
   
2. import java.util.Arrays;
   
3. import java.util.Collection;
   
4. import java.util.Iterator;
   
5. 
   
6. public class Main {
   
7.   public static void main(String[] args) {
   
8.     int[] first = new int[] { 0, 1, 2 };
   
9.     int[] second = new int[] { 4, 5, 6 };
   
10. 
    
11.     MyStream.concat(MyStream.of(first).map(x -> (int)x + 1),
    
12.                     MyStream.of(second))                        // 1, 2, 3, 4, 5, 6
    
13.             .map(x -> (int)x * (int)x)                          // 1, 4, 9, 16, 25, 36
    
14.             .filter(x -> (int)x > 5 && (int)x < 30)             // 9, 16, 25
    
15.             .reduce(0, (x, y) -> (int)x + (int)y)               // 50
    
16.             .each(x -> System.out.println("sum = " + x));       // sum = 50
    
17.   }
    
18. }
    
19. 
    
20. // map()方法的参数类型：一元函数
    
21. interface UnaryOperation {
    
22.   public Object apply(Object operand);
    
23. }
    
24. 
    
25. // reduce()方法的参数类型：二元函数
    
26. interface BinaryOperation {
    
27.   public Object apply(Object lhs, Object rhs);
    
28. }
    
29. 
    
30. // filter()方法的参数类型：一元谓词
    
31. interface UnaryPredicate {
    
32.   public boolean apply(Object operand);
    
33. }
    
34. 
    
35. // each()方法的参数类型
    
36. interface Consumer {
    
37.   public void apply(Object operand);
    
38. }
    
39. 
    
40. // 不考虑性能的简单实现
    
41. @SuppressWarnings("rawtypes")
    
42. abstract class MyStream implements Iterable {
    
43.   public MyStream filter(UnaryPredicate predicate) {
    
44.     ArrayList<Object> results = new ArrayList<Object>();
    
45.     for (Object o : this) {
    
46.       if (predicate.apply(o)) {
    
47.         results.add(o);
    
48.       }
    
49.     }
    
50.     return of(results);
    
51.   }
    
52. 
    
53.   public MyStream map(UnaryOperation operation) {
    
54.     ArrayList<Object> results = new ArrayList<Object>();
    
55.     for (Object o : this) {
    
56.       results.add(operation.apply(o));
    
57.     }
    
58.     return of(results);
    
59.   }
    
60. 
    
61.   public MyStream reduce(Object init, BinaryOperation operation) {
    
62.     Object accumulated = init;
    
63.     for (Object o : this) {
    
64.       if (accumulated == null) {
    
65.         accumulated = o;
    
66.       } else {
    
67.         accumulated = operation.apply(accumulated, o);
    
68.       }
    
69.     }
    
70.     ArrayList<Object> results = new ArrayList<Object>();
    
71.     if (accumulated != null) {
    
72.       results.add(accumulated);
    
73.     }
    
74.     return of(results);
    
75.   }
    
76. 
    
77.   public MyStream reduce(BinaryOperation operation) {
    
78.     return reduce(null, operation);
    
79.   }
    
80. 
    
81.   public void each(Consumer consumer) {
    
82.     for (Object o : this) {
    
83.       consumer.apply(o);
    
84.     }
    
85.   }
    
86. 
    
87. 
    
88.   public static MyStream concat(MyStream a, MyStream b) {
    
89.     ArrayList<Object> results = new ArrayList<Object>();
    
90.     for (Object o : a) {
    
91.       results.add(o);
    
92.     }
    
93.     for (Object o : b) {
    
94.       results.add(o);
    
95.     }
    
96.     return of(results);
    
97.   }
    
98. 
    
99.   public static MyStream of(int[] data) {
    
100.     ArrayList<Integer> results = new ArrayList<Integer>();
     
101.     for (int i = 0; i < data.length; ++i) {
     
102.       results.add(data[i]);
     
103.     }
     
104.     return of(results);
     
105.   }
     
106. 
     
107.   public static MyStream of(Object[] data) {
     
108.     return new ArrayStream(data);
     
109.   }
     
110. 
     
111.   public static MyStream of(Collection data) {
     
112.     return new CollectionStream(data);
     
113.   }
     
114. }
     
115. 
     
116. class ArrayStream extends MyStream {
     
117.   private Object[] data;
     
118. 
     
119.   public ArrayStream(Object[] data) {
     
120.     this.data = data;
     
121.   }
     
122. 
     
123.   @Override
     
124.   public Iterator<Object> iterator() {
     
125.     return Arrays.asList(data).iterator();
     
126.   }
     
127. }
     
128. 
     
129. @SuppressWarnings("rawtypes")
     
130. class CollectionStream extends MyStream {
     
131.   private Collection data;
     
132. 
     
133.   CollectionStream(Collection data) {
     
134.     this.data = data;
     
135.   }
     
136. 
     
137.   @Override
     
138.   public Iterator iterator() {
     
139.     return data.iterator();
     
140.   }
     
141. }
     

复制代码