拆轮子系列：go的流式处理库go

由于某些原因需要做点骚操作，需要用到类似于flink中的滑动窗口这样的功能，此时我打开了awesome-go上去瞄瞄，看看有没有开源的实现，然后就发现了这篇文章的主角go-streams这个库。

go-streams是什么

一个流式处理库，提供一种简单，简洁的方式构建数据处理的管道。

pipeline流水线处理的需求点有哪些

pipeline流水线处理模式对于解决问题的建模是将任务分为多个依次执行的处理步骤，步骤间通过传输数据相关联：一个步骤的输出是下一个步骤的输入。可以用生产者消费者模型来理解，上游生产数据，下游消费数据，golang当中的goruntine和channel非常适合实现，每一步处理都开一个或多个goruntine，通过channel来连接不同的步骤的goruntine。

需求点

支持不同的输入数据源。例如：kafka, redis, 文本，网络等等
最终结果能够以各种方式存储或者显示。例如：kafka, redis, 文本，网络等等
支持常见的数据操作，例如：filter, map等等
支持数据流关系操作，合并（即多个步骤的输出，往同一个步骤输入），拆分（一个步骤的往多个步骤输入相同数据）

简单的例子

一般开源项目都会有使用的例子，一般目录是examples，直接去到example目录的std目录下找到下面的代码。

package main

import (
    "strconv"
    "time"

    ext "github.com/reugn/go-streams/extension"
    "github.com/reugn/go-streams/flow"
)

type message struct {
    Msg string
}

func (msg *message) String() string {
    return msg.Msg
}

func main() {
    // 构建输入源：该输入源每隔一段时间，输出当前时间的字符串
    source := ext.NewChanSource(tickerChan(time.Second * 1))

    // 对数据进行处理：时间字符后面加上"-UTC"
    flow := flow.NewMap(mapp, 1)

  // 结果输出：将结果输出到标准输出
    sink := ext.NewStdoutSink()

    // 将流式处理各个步骤串联起来
    source.Via(flow).To(sink)

    select {}
}

var mapp = func(in interface{}) interface{} {
    msg := in.(*message)
    msg.Msg += "-UTC"
    return msg
}

func tickerChan(repeat time.Duration) chan interface{} {
    ticker := time.NewTicker(repeat)
    oc := ticker.C
    nc := make(chan interface{})
    go func() {
        for range oc {
            nc <- &message{strconv.FormatInt(time.Now().UTC().UnixNano(), 10)}
        }
    }()
    return nc
}

go-streams是如何抽象并实现相关需求的

支持不同的输入数据源
为了支持不同的输入源，go-streams抽象出来Source接口，不同的输入源，还官方实现了常见的输入源：文本，kafka，redis，channel

最终结果能够以各种方式存储或者显示
go-streams抽象出来Sink接口，不同的输出形式，官网也实现了常见的形式：文本，kafka，redis，channel

支持常见的数据操作 & 支持数据流关系操作
go-streams抽象出来Flow接口，官方实现了常见的操作。

Map
FlatMap
Filter
PassThrough
Split
FanOut
Merge
Throttler
SlidingWindow（这个是我想要的滑动窗口）
TumblingWindow

代码欣赏

类体系构建

package streams

//Inlet - In port interface
type Inlet interface {
    In() chan<- interface{}
}

//Outlet - Out port interface
type Outlet interface {
    Out() <-chan interface{}
}

//Source adaptor
type Source interface {
    // 输入源只会输出数据，所以只有Outlet
    Outlet

    // 
  // 这样就是可以实现连接
    Via(Flow) Flow
}

//Flow - Stream transformation interface
// 各种各样的数据操作，
// 有输入，可能会输出到下一个处理步骤，也可能是输出到最终结果里面
type Flow interface {
    Inlet
    Outlet

    // 这个方法输入一个Flow接口，输出也是一个flow接口，这样就可以实现链式调用
  // 例子：source.Via(flow).To(sink)，给予一般正常的流式处理的编写形式一致的体验
    Via(Flow) Flow
    To(Sink)

}

//Sink adaptor
type Sink interface {
    Inlet
}

滑动窗口Flow接口的实现

什么是滑动窗口

举一个简单的应用例子，每隔30s统计最近10分钟活跃用户数。
窗口：以固定时间来收集数据，例如：10分钟活跃用户数
滑动：以固定的时间向前滑动，例如：每隔30s统计一次

抽象出来的数据结构

窗口数据的存储是利用按照时间排序优先级的优先级队列来存储。这里需要注意的是，按照时间生产的位置不同可以分位: 事件生成时间，事件接入时间，事件处理时间。由于消息到达的顺序可能会乱序，因此需要通过时间排序的优先级队列，来解决乱序问题，而切片无法满足需求。

type SlidingWindow struct {
    sync.Mutex

    // 窗口时间
    size               time.Duration

    // 滑动时间
    slide              time.Duration

    // 一个按照时间排序的优先级队列，
    queue              *PriorityQueue
    in                 chan interface{}
    out                chan interface{}

    // 通过timestampExtractor指定对应的时间类型，默认是事件处理时间。
    timestampExtractor func(interface{}) int64
    closed             bool
}

关键方法

整体流程：滑动找到对应窗口数据 —> 优先级队列删除不会再使用的数据 —> 输出窗口数据

func (sw *SlidingWindow) emit() {
    for {
        select {
        // 直接：<-time.After(sw.slide)，不是推荐的用法，一不小心容易导致内存泄露。
        case <-time.After(sw.slide):
            sw.Lock()
            //build window slice and send to out chan
            var slideUpperIndex, windowBottomIndex int
            now := streams.NowNano()

            // 获取优先级队列的长度
            windowUpperIndex := sw.queue.Len()

            // 下次一次的windowBottomTime位置
            slideUpperTime := now - sw.size.Nanoseconds() + sw.slide.Nanoseconds()

            
            windowBottomTime := now - sw.size.Nanoseconds()
            for i, item := range *sw.queue {
                if item.epoch < windowBottomTime {
                    windowBottomIndex = i
                }
                if item.epoch > slideUpperTime {
                    slideUpperIndex = i
                    break
                }
            }

            // 获取当前时间窗口的数据
            windowSlice := extract(sw.queue.Slice(windowBottomIndex, windowUpperIndex))
            if windowUpperIndex > 0 {
                // 删除不再使用的数据，只保留：滑动的时间对应的位置--最后的位置
                s := sw.queue.Slice(slideUpperIndex+1, windowUpperIndex)
                //reset queue
                sw.queue = &s
                heap.Init(sw.queue)
            }

            sw.Unlock()
            if sw.closed {
                break
            }

            // 输出数据
            if len(windowSlice) > 0 {
                sw.out <- windowSlice
            }
        }
    }
}

为什么说直接在case当中<-time.After(sw.slide)容易导致内存泄露

for {
      select {
      case x := <- ch:
          println(x)
            
            // 每次for循环都新建一个ticker, 系统要3分钟后，才会关闭创建的timer
            // 也就是说每次都会泄露一个channel
      case <- time.After(3 * time.Minute):
          println(time.Now().Unix())
      }
  }

良好的做法是

tick := time.NewTicker(3 * time.Minute)
defer tick.Stop()

for {
    select {
    case <-tick.C:
        fmt.Println("hi")
    case <-c:
        fmt.Println("do")
    }
}