go-streams 并发 示范

以下是一个使用 go-streams 进行并发处理的示范代码：

package main

import (
    "fmt"
    "github.com/reugn/go-streams/flow"
    "github.com/reugn/go-streams/runner"
)

func main() {
    // 构建处理器流程
    stream := flow.NewStream(
        flow.NewSliceSource([]interface{}{1, 2, 3, 4, 5}),
        flow.NewMap(func(i interface{}) interface{} {
            return i.(int) * 2
        }),
        flow.NewFilter(func(i interface{}) bool {
            return i.(int) > 5
        }),
        flow.NewReduce(func(a interface{}, b interface{}) interface{} {
            return a.(int) + b.(int)
        }),
        flow.NewSink(func(v interface{}) {
            fmt.Println(v)
        }),
    )

    // 使用4个协程并发执行处理器流程
    runner := runner.NewParallelRunner(stream, 4)
    runner.Start()
    runner.AwaitTermination()
}

这个示范代码中，我们首先构建了一个处理器流程，包含了 SliceSource、Map、Filter、Reduce 和 Sink 五个处理器。然后，我们创建了一个并发运行器（ParallelRunner），将处理器流程和并发数（4）作为参数传入并启动运行器。最后，我们等待运行器的终止即可。

通过并发执行处理器流程，我们可以大大提高数据处理的效率，尤其是在处理大量数据时，效果更加显著。