fx

fx 是一个完备的流式处理组件。 和 MapReduce 类似的，fx 也存在并发处理的函数：Parallel(fn, options)。但同时它也不只有并发处理。From(chan) ，Map(fn)，Filter(fn)，Reduce(fn) 等，从数据源读取成流，到处理流数据，最后聚合流数据。是不是有点像 Java Lambda ，如果你之前是 Java 开发者，看到这也就明白整个基本设计。

整体API

还是从整体上概览 fx 到底是怎么构建的：

被标注的部分就是整个 fx 最重要的部分：

1. 由 From(fn) 这类API，产生数据流 Stream
2. 对Stream 转换，聚合，求值的API集合

所以列举出目前支持的 Stream API：

不再生成新的 stream，做最后的求值操作：

如何使用

result := make(map[string]string)
fx.From(func(source chan<- interface{}) {
  for _, item := range data {
    source <- item
  }
}).Walk(func(item interface{}, pipe chan<- interface{}) {
  each := item.(*model.ClassData)
  class, err := l.rpcLogic.GetClassInfo()
  if err != nil {
    l.Errorf("get class %s failed: %s", each.ClassId, err.Error())
    return
  }
  students, err := l.rpcLogic.GetUsersInfo(class.ClassId)
  if err != nil {
    l.Errorf("get students %s failed: %s", each.ClassId, err.Error())
    return
  }
  pipe <- &classObj{
    classId: each.ClassId
    studentIds: students
  }
}).ForEach(func(item interface{}) {
    o := item.(*classObj)
    result[o.classId] = o.studentIds
})

1. From() 从一个 slice 生成 stream
2. Walk() 接收与一个 stream ，对流中每个 ele 转换重组，生成新的 stream
3. 最后由 求值操作 把 stream 输出（fmt.Println），存储（map,slice），持久化（db操作）

简要分析

fx 中的函数命名语义化，开发者只需要知道业务逻辑需要什么样的转换，调用与之匹配的函数即可。

所以这里只简要分析几个比较典型的函数。

Walk()

Walk() 在整个 fx 被多个函数当成底层实现，Map(), Filter() 等。

所以本质就是：Walk() 负责并发将传进来的函数作用在 输入流 的每个 ele，并 生成新的 stream。

跟到源码，分成两个子函数：自定义 worker 数，默认 worker 数

// 自定义 workder 数
func (p Stream) walkLimited(fn WalkFunc, option *rxOptions) Stream {
    pipe := make(chan interface{}, option.workers)
    go func() {
        var wg sync.WaitGroup
    // channel<- 如果达到设定workers数，channel阻塞，从而达到控制并发数。
    // 简易的 goroutine pool
        pool := make(chan lang.PlaceholderType, option.workers)
        for {
      // 每一次for循环都会开启一个goroutine。如果达到workers数，从而阻塞
            pool <- lang.Placeholder
            item, ok := <-p.source
            if !ok {
                <-pool
                break
            }
            // 使用 waitgroup 保证任务完成的完整性
            wg.Add(1)
            threading.GoSafe(func() {
                defer func() {
                    wg.Done()
          // 归还
                    <-pool
                }()
                fn(item, pipe)
            })
        }
        wg.Wait()
        close(pipe)
    }()
    return Range(pipe)
}

• 使用 有缓冲channel 做并发队列，限制并发数
• waitgroup 保证任务完成的完整性

另外一个 walkUnlimited()：也使用了 waitgroup 做并发控制，因为没有自定义并发数限制，所以也就没有另外一个 channel 做并发数控制。

Tail()

介绍这个主要是里面运用了 ring 这个双向链表，其中的简单算法还是很有意思的。

func (p Stream) Tail(n int64) Stream {
    source := make(chan interface{})
    go func() {
        ring := collection.NewRing(int(n))
    // “顺序”插入，源的顺序和ring的顺序一致
        for item := range p.source {
            ring.Add(item)
        }
    // 取出 ring 中全部的 item
        for _, item := range ring.Take() {
            source <- item
        }
        close(source)
    }()
    return Range(source)
}

至于为什么 Tail() 可以做到把源的后n个取出来，这个就留给大家去细品了。这里给出我的理解：

[!TIP] 假设有以下这个场景，Tail(5)

• stream size ：7
• ring size：5

这里可以使用把环状链表拉开的方式，环转线，此时以全部长度划分对称轴，翻转多余的元素，之后的元素就是 Tail(5) 需要的部分了。

[!TIP] 这里采用图的方式更清晰的表现，不过代码大家也要看看。算法要考的

Stream Transform Design

分析整个 fx ，会发现整体设计遵循一个设计模版：

func (p Stream) Transform(fn func(item interface{}) interface{}) Stream {
    // make channel
    source := make(chan interface{})
    // goroutine worker
    go func() {
        // tranform
        for item := range p.source {
            ...
            source <- item
            ...
        }
        ...
        // 关闭输入，但是依然可以从这个 stream 输出。防止内存泄漏
        close(source)
    }()
    // channel -> stream
    return Range(source)
}

• channel 作为流的容器
• 开 goroutine 对 source 做转换，聚合，输送到 channel
• 处理完毕，close(outputStream)
• channel -> stream

总结

到这就把 fx 基本介绍完了，如果你对其他API源码感兴趣，可以跟着上面的 API 列表挨个读一读。

同时也建议大家把 java stream 的API大致看看，对这种 stream call 理解可以更加深 。

同时在 go-zero 中还有很多实用的组件工具，用好工具对于提升服务性能和开发效率都有很大的帮助，希望本篇文章能给大家带来一些收获。

参考资料

• go-zero
• Java Stream 详解
• Java 8中Stream API