C++ 实现 Java 风格 Stream API 的思路与代码

将一个 Vector 转换成流：

static Flow<T> of(std::vector<T> vector) { 
    return Flow<T>([=](const std::function<void(T)> &c) { 
        for (const auto &item: vector) { 
            c(item); 
        } 
    }); 
} 

map

对于 map 其实就是消费一个流，并且产生一个新的流。

template<typename R> Flow<R> map(std::function<R(T)> dataMapFun) { 
    Flow<T> mapFLow = Flow([=](std::function<void(R)> c) { 
        // 定义了上一个流消费逻辑，消费的同时转换数据，产生下一个流
        this->consume([=](T data) { 
            c(dataMapFun(data)); 
        }); 
    }); 
    return mapFLow; 
} 

记住我们对一个 consume 函数的定义，调用入参的回调函数即是生产数据。

flatmap

template<typename R> Flow<R> flatMap(std::function<Flow<R>(T)> function) { 
    return Flow([=](std::function<void(R)> c) { 
        consume([=](T data) { 
            Flow<R> flow = function(data); 
            flow.consume(c); 
        }); 
    }); 
} 

takeWhile

Flow<T> takeWhile(std::function<bool(T)> predicate) { 
    return Flow( [=](std::function<void(T)> c) { 
        consumeTillStop([=](T data) { 
            if (predicate(data)) { 
                stop(); 
            } 
            c(data); 
        }); 
    } ); 
} 
void stop() { throw StopException{}; } 
void consumeTillStop(std::function<void(T)> consumer) { 
    try { 
        consume(consumer); 
    } catch (StopException exception) { 
    } 
} 

因为都是回调函数，我们只能通过异常通知生产者停止生产。

filter

Flow<T> filter(std::function<bool(T)> predicate) { 
    return Flow( [=](std::function<void(T)> c) { 
        consume([=](T data) { 
            if (predicate(data)) { 
                c(data); 
            } 
        }); 
    } ); 
} 

sorted

Flow<T> sorted(std::function<bool(T, T)> comparator) { 
    auto list = this->toVector(); 
    std::sort(list.begin(), list.end(), comparator); 
    return Flow::of(list); 
} 

toVector

类似 java 中 toList，这里是最终的操作，不要产生一个新的流了。

std::vector<T> toVector() { 
    std::vector<T> list; 
    this->consume([&list](T data) { 
        list.push_back(data); 
    }); 
    return list; 
} 

扩展

上面只给出了部分 api 的实现，按照这个思路 stream 中所有 api 都是可以实现的，上面还给出了 java9 中才支持的 takeWhile。

上面也只给出了 Vector 转换为流，只要牢记我们那个关键函数的定义 consume：当流生产出数据时就调用入参的回调函数，那么就可以实现所有集合的流式操作。

那么它仅限于此么？当然不是。

并行流

并发多线程调用回调函数，那么这个流就变成了并发的了。

文件流

当读出文件时，就调用回调函数，那么它就变成了文件流，而且是基于回调函数的，不必读取所有的文件内容，读一部分，处理一部分，占用内存十分的小。

二元流

看看下面这个函数的定义，它又变成了二元流了，当然 N 元的也不是不可以。

template<typename T1,typename T2> class Flow { 
protected: 
    std::function<void(std::function<void(T1,T2)>)> consume; 
} 

业务流

最后回到生成者消费者模型，仔细想想也满足上述模型，而且是背压的，少了一个存放数据队列。

总结

我们通过级联回调函数的方式实现了 java 中 stream 相关 api，比起 java jdk 的实现要简单很多，但是该有的功能都有了。

这些代码的实现正是业务逻辑与控制逻辑分离的体现，api 的调用者只关心逻辑，不关心怎么去控制。比如 filter 你只需要告诉我过滤的条件是什么，怎么样过滤不需要关心。

也是封装易变点思想的体现，通过上述 api 的封装，对于 map，filter 等等操作，不用重复去写 for 循环，if 语句，只需要告诉 api 的逻辑是什么，也就是传入的 func。

也是函数式编程思想的体现，无状态的，函数作为参数传递，惰性求值和并性处理等。

代码 demo

#include <vector> 
#include <list> 
#include <set> 
#include <map> 
#include <functional> 
#include <iostream>

/** 
 * c++ 版本流的实现，通过回调函数的方式实现类似 java 中 stream 的 api，目前是 demo 版本，支持部分功能，看需要可能会支持并发流，二元流等功能 
 *
 * 由于是回调函数实现，天然是懒加载，被压的方式 
 *
 * @tparam T 流的类型 
 */ 
template<typename T> class Flow { 
private: 
    void stop() { throw StopException{}; } 
    void consumeTillStop(std::function<void(T)> consumer) { 
        try { 
            consume(consumer); 
        } catch (StopException exception) { 
        } 
    } 
protected: 
    std::function<void(std::function<void(T)>)> consume; 
public: 
    explicit Flow(const std::function<void(std::function<void(T)>)> c) : consume(c) {} 
    using StopException = std::exception; 
    
    void forEach(std::function<void(T)> consumer) { 
        this->consume(consumer); 
    } 
    
    template<typename R> Flow<R> map(std::function<R(T)> dataMapFun) { 
        Flow<T> mapFLow = Flow([=](std::function<void(R)> c) { 
            consume([=](T data) { 
                c(dataMapFun(data)); 
            }); 
        }); 
        return mapFLow; 
    } 
    
    template<typename R> Flow<R> flatMap(std::function<Flow<R>(T)> function) { 
        return Flow([=](std::function<void(R)> c) { 
            consume([=](T data) { 
                Flow<R> flow = function(data); 
                flow.consume(c); 
            }); 
        }); 
    } 
    
    Flow<T> takeWhile(std::function<bool(T)> predicate) { 
        return Flow( [=](std::function<void(T)> c) { 
            consumeTillStop([=](T data) { 
                if (predicate(data)) { 
                    stop(); 
                } 
                c(data); 
            }); 
        } ); 
    } 
    
    Flow<T> filter(std::function<bool(T)> predicate) { 
        return Flow( [=](std::function<void(T)> c) { 
            consume([=](T data) { 
                if (predicate(data)) { 
                    c(data); 
                } 
            }); 
        } ); 
    } 
    
    Flow<T> dropWhile(std::function<bool(T)> predicate) { 
        return Flow( [=](std::function<void(T)> c) { 
            bool drop = false; 
            consume([=, &drop](T data) { 
                if (drop) { 
                    c(data); 
                } 
                if (!drop) { 
                    drop = predicate(data); 
                } 
            }); 
        } ); 
    } 
    
    Flow<T> skip(int n) { 
        return Flow( [=](std::function<void(T)> c) { 
            int count = n; 
            consume([=, &count](T data) { 
                if (count <= 0) { 
                    c(data); 
                } else { 
                    count--; 
                } 
            }); 
        } ); 
    } 
    
    Flow<T> peek(std::function<void(T)> consumer) { 
        return Flow( [=](std::function<void(T)> c) { 
            consume([=](T data) { 
                consumer(data); 
                c(data); 
            }); 
        } ); 
    } 
    
    Flow<T> sorted(std::function<bool(T, T)> comparator) { 
        auto list = this->toVector(); 
        std::sort(list.begin(), list.end(), comparator); 
        return Flow::of(list); 
    } 
    
    Flow<T> sorted() { 
        auto list = this->toVector(); 
        std::sort(list.begin(), list.end()); 
        return Flow::of(list); 
    } 
    
    std::vector<T> toVector() { 
        std::vector<T> list; 
        this->consume([&list](T data) { 
            list.push_back(data); 
        }); 
        return list; 
    } 
    
    template<typename K> std::unordered_map<K, std::vector<T>> groupBy(std::function<K(T)> keyFun) { 
        std::unordered_map<K, std::vector<T>> map; 
        consume([=, &map](T data) { 
            K key = keyFun(data); 
            auto iter = map.find(key); 
            if (iter != map.end()) { 
                iter->second.push_back(data); 
            } else { 
                std::vector<T> vector; 
                vector.push_back(data); 
                map.insert(std::make_pair(key, vector)); 
            } 
        }); 
        return map; 
    } 
    
    std::set<T> toSet() { 
        std::set<T> set; 
        this->consume([&set](T data) { 
            set.insert(data); 
        }); 
        return set; 
    } 
    
    static Flow<T> of(std::vector<T> vector) { 
        return Flow<T>([=](const std::function<void(T)> &c) { 
            for (const auto &item: vector) { 
                c(item); 
            } 
        }); 
    } 
}; 

std::vector<std::string> vec; 
vec.push_back("12_3"); 
vec.push_back("12"); 
vec.push_back("11"); 
vec.push_back("a"); 
vec.push_back("b"); 
Flow<std::string> strFlow = Flow<std::string>::of(vec); 
std::cout << "demo1<<<<<<<<<<<<<<<<<<:" << std::endl; 
std::vector<std::string> mapVec = strFlow 
.map<std::string>([=](const std::string &data) { 
    return data + "123"; 
}) 
.peek([=](const std::string &data) { 
    std::cout << "peek:" + data << std::endl; 
}) 
.skip(3) 
.toVector(); 

tips

为常用函数定义别名，提升可读性。

using Consumer = std::function<void(T)>; 
using Function = std::function<T(T)>; 
using Predicate = std::function<bool(T)>; 
using Comparator = std::function<bool(T,T)>;