简介
ReactiveX，简称为 Rx，是一个异步编程的 API。与 callback（回调）、promise（JS 提供这种形式）和 deferred（Python 的 twisted 网络编程库就是应用这种形式）这些异步编程形式有所不同，Rx 是基于事件流的。这里的事件能够是零碎中产生或变动的任何货色，在代码中咱们个别用对象示意。在 Rx 中，事件流被称为 Observable（可察看的）。事件流须要被 Observer（观察者）解决才有意义。设想一下，咱们日常作为一个 Observer，一个重要的工作就是察看 BUG 的事件流。每次发现一个 BUG，咱们都须要去解决它。
Rx 仅仅只是一个 API 标准的定义。Rx 有多种编程语言实现，RxJava/RxJS/Rx.NET/RxClojure/RxSwift。RxGo 是 Rx 的 Go 语言实现。借助于 Go 语言简洁的语法和弱小的并发反对（goroutine、channel），Rx 与 Go 语言的联合十分完满。
pipelines (官网博客：https://blog.golang.org/pipelines)是 Go 根底的并发编程模型。其中蕴含，fan-in——多个 goroutine 产生数据，一个goroutine 解决数据，fan-out——一个 goroutine 产生数据，多个 goroutine 解决数据，fan-inout——多个 goroutine 产生数据，多个 goroutine 解决数据。它们都是通过 channel 连贯。RxGo 的实现就是基于 pipelines 的理念，并且提供了不便易用的包装和弱小的扩大。
疾速应用
本文代码应用 Go Modules。
创立目录并初始化：
$ mkdir rxgo && cd rxgo$ go mod init github.com/darjun/go-daily-lib/rxgo装置rxgo库：
$ go get -u github.com/reactivex/rxgo/v2编码：
package mainimport ( "fmt" "github.com/reactivex/rxgo/v2")func main() { observable := rxgo.Just(1, 2, 3, 4, 5)() ch := observable.Observe() for item := range ch { fmt.Println(item.V) }}应用 RxGo 的个别流程如下：
- 应用相干的 Operator 创立 Observable，Operator 就是用来创立 Observable 的。这些术语都比拟难贴切地翻译，而且英文也很好懂，就不强行翻译了；
- 两头各个阶段能够应用过滤操作筛选出咱们想要的数据，应用转换操作对数据进行转换；
- 调用 Observable 的
Observe()办法，该办法返回一个<- chan rxgo.Item。而后for range遍历即可。
GitHub 上一张图很形象地描述了这个过程：
- 首先应用
Just创立一个仅有若干固定数据的 Observable； - 应用
Map()办法执行转换（将圆形转为方形）； - 应用
Filter()办法执行过滤（过滤掉黄色的方形）。
看懂了这张图片，就能理解 RxGo 工作的根本流程了。
下面是简略的示例，没有过滤、转换操作的应用。
运行：
$ go run main.go 12345对于下面的示例，须要留神：
Just应用柯里化（currying）让它能够在第一个参数中承受多个数据，在第二个参数中承受多个选项定制行为。柯里化是函数化编程的思维，简略来说就是通过在函数中返回函数，以此来缩小每个函数的参数个数。例如：
func add(value int) func (int) int { return func (a int) int { return value + a }}fmt.Prinlnt(add(5)(10)) // 15因为 Go 不反对多个可变参数，Just通过柯里化曲折地实现了这个性能：
// rxgo/factory.gofunc Just(items ...interface{}) func(opts ...Option) Observable { return func(opts ...Option) Observable { return &ObservableImpl{ iterable: newJustIterable(items...)(opts...), } }}实际上rxgo.Item还能够蕴含谬误。所以在应用时，咱们应该做一层判断：
func main() { observable := rxgo.Just(1, 2, errors.New("unknown"), 3, 4, 5)() ch := observable.Observe() for item := range ch { if item.Error() { fmt.Println("error:", item.E) } else { fmt.Println(item.V) } }}运行：
$ go run main.go 12error: unknown345咱们应用item.Error()查看是否呈现谬误。而后应用item.V拜访数据，item.E拜访谬误。
除了应用for range之外，咱们还能够调用 Observable 的ForEach()办法来实现遍历。ForEach()承受 3 个回调函数：
NextFunc：类型为func (v interface {})，解决数据；ErrFunc：类型为func (err error)，处理错误；CompletedFunc：类型为func ()，Observable 实现时调用。
有点Promise那味了。应用ForEach()，能够将下面的示例改写为：
func main() { observable := rxgo.Just(1, 2, 3, 4, 5)() <-observable.ForEach(func(v interface{}) { fmt.Println("received:", v) }, func(err error) { fmt.Println("error:", err) }, func() { fmt.Println("completed") })}运行：
$ go run main.go received: 1received: 2received: 3received: 4received: 5completedForEach()实际上是异步执行的，它返回一个接管告诉的 channel。当 Observable 数据发送结束时，该 channel 会敞开。所以如果要期待ForEach()执行实现，咱们须要应用<-。下面的示例中如果去掉<-，可能就没有输入了，因为主 goroutine 完结了，整个程序就退出了。
创立 Observable
下面应用最简略的形式创立 Observable：间接调用Just()办法传入一系列数据。上面再介绍几种创立 Observable 的形式。
Create
传入一个[]rxgo.Producer的切片，其中rxgo.Producer的类型为func(ctx context.Context, next chan<- Item)。咱们能够在代码中调用rxgo.Of(value)生成数据，rxgo.Error(err)生成谬误，而后发送到next通道中：
func main() { observable := rxgo.Create([]rxgo.Producer{func(ctx context.Context, next chan<- rxgo.Item) { next <- rxgo.Of(1) next <- rxgo.Of(2) next <- rxgo.Of(3) next <- rxgo.Error(errors.New("unknown")) next <- rxgo.Of(4) next <- rxgo.Of(5) }}) ch := observable.Observe() for item := range ch { if item.Error() { fmt.Println("error:", item.E) } else { fmt.Println(item.V) } }}当然，分成两个rxgo.Producer也是一样的成果：
observable := rxgo.Create([]rxgo.Producer{func(ctx context.Context, next chan<- rxgo.Item) { next <- rxgo.Of(1) next <- rxgo.Of(2) next <- rxgo.Of(3) next <- rxgo.Error(errors.New("unknown")) }, func(ctx context.Context, next chan<- rxgo.Item) { next <- rxgo.Of(4) next <- rxgo.Of(5)}})FromChannel
FromChannel能够间接从一个已存在的<-chan rxgo.Item对象中创立 Observable：
func main() { ch := make(chan rxgo.Item) go func() { for i := 1; i <= 5; i++ { ch <- rxgo.Of(i) } close(ch) }() observable := rxgo.FromChannel(ch) for item := range observable.Observe() { fmt.Println(item.V) }}留神：
通道须要手动调用close()敞开，下面Create()办法外部rxgo主动帮咱们执行了这个步骤。
Interval
Interval以传入的工夫距离生成一个无穷的数字序列，从 0 开始：
func main() { observable := rxgo.Interval(rxgo.WithDuration(5 * time.Second)) for item := range observable.Observe() { fmt.Println(item.V) }}下面的程序启动后，第 5s 输入 0，第 10s 输入 1，...，而且不会进行。
咱们能够用time.Ticker实现雷同的性能：
func main() { t := time.NewTicker(5 * time.Second) var count int for range t.C { fmt.Println(count) count++ }}Range
Range能够生成一个范畴内的数字：
func main() { observable := rxgo.Range(0, 3) for item := range observable.Observe() { fmt.Println(item.V) }}下面代码顺次输入 0，1，2，3。
Repeat
在已存在的 Observable 对象上调用Repeat，能够实现每隔指定工夫，反复一次该序列，一共反复指定次数：
func main() { observable := rxgo.Just(1, 2, 3)().Repeat( 3, rxgo.WithDuration(1*time.Second), ) for item := range observable.Observe() { fmt.Println(item.V) }}运行下面的代码，立刻输入 1，2，3，而后期待 1s，又输入一次 1，2，3，而后又期待 1s，最初又输入一次 1，2，3。
Start
能够给Start办法传入[]rxgo.Supplier作为参数，它能够蕴含任意数量的rxgo.Supplier类型。rxgo.Supplier的底层类型为：
// rxgo/types.govar Supplier func(ctx context.Context) rxgo.ItemObservable 外部会顺次调用这些rxgo.Supplier生成rxgo.Item：
func Supplier1(ctx context.Context) rxgo.Item { return rxgo.Of(1)}func Supplier2(ctx context.Context) rxgo.Item { return rxgo.Of(2)}func Supplier3(ctx context.Context) rxgo.Item { return rxgo.Of(3)}func main() { observable := rxgo.Start([]rxgo.Supplier{Supplier1, Supplier2, Supplier3}) for item := range observable.Observe() { fmt.Println(item.V) }}Observable 分类
依据数据在何处生成，Observable 被分为 Hot 和 Cold 两种类型（类比热启动和冷启动）。数据在其它中央生成的被成为 Hot Observable。相同，在 Observable 外部生成数据的就是 Cold Observable。
应用下面介绍的办法创立的实际上都是 Hot Observable。
func main() { ch := make(chan rxgo.Item) go func() { for i := 0; i < 3; i++ { ch <- rxgo.Of(i) } close(ch) }() observable := rxgo.FromChannel(ch) for item := range observable.Observe() { fmt.Println(item.V) } for item := range observable.Observe() { fmt.Println(item.V) }}下面创立的是 Hot Observable。然而有个问题，第一次Observe()耗费了所有的数据，第二个就没有数据输入了。
而 Cold Observable 就不会有这个问题，因为它创立的流是独立于每个观察者的。即每次调用Observe()都创立一个新的 channel。咱们应用Defer()办法创立 Cold Observable，它的参数与Create()办法一样。
func main() { observable := rxgo.Defer([]rxgo.Producer{func(_ context.Context, ch chan<- rxgo.Item) { for i := 0; i < 3; i++ { ch <- rxgo.Of(i) } }}) for item := range observable.Observe() { fmt.Println(item.V) } for item := range observable.Observe() { fmt.Println(item.V) }}输入：
$ go run main.go012012可连贯的 Observable
可连贯的（Connectable）Observable 对一般的 Observable 进行了一层组装。调用它的Observe()办法时并不会立即产生数据。应用它，咱们能够等所有的观察者都准备就绪了（即调用了Observe()办法）之后，再调用其Connect()办法开始生成数据。咱们通过两个示例比拟应用一般的 Observable 和可连贯的 Observable 有何不同。
一般的：
func main() { ch := make(chan rxgo.Item) go func() { for i := 1; i <= 3; i++ { ch <- rxgo.Of(i) } close(ch) }() observable := rxgo.FromChannel(ch) observable.DoOnNext(func(i interface{}) { fmt.Printf("First observer: %d\n", i) }) time.Sleep(3 * time.Second) fmt.Println("before subscribe second observer") observable.DoOnNext(func(i interface{}) { fmt.Printf("Second observer: %d\n", i) }) time.Sleep(3 * time.Second)}上例中咱们应用DoOnNext()办法来注册观察者。因为DoOnNext()办法是异步执行的，所以为了期待后果输入，在最初减少了一行time.Sleep。运行：
$ go run main.goFirst observer: 1First observer: 2First observer: 3before subscribe second observer由输入能够看出，注册第一个观察者之后就开始产生数据了。
咱们通过在创立 Observable 的办法中指定rxgo.WithPublishStrategy()选项就能够创立可连贯的 Observable：
func main() { ch := make(chan rxgo.Item) go func() { for i := 1; i <= 3; i++ { ch <- rxgo.Of(i) } close(ch) }() observable := rxgo.FromChannel(ch, rxgo.WithPublishStrategy()) observable.DoOnNext(func(i interface{}) { fmt.Printf("First observer: %d\n", i) }) time.Sleep(3 * time.Second) fmt.Println("before subscribe second observer") observable.DoOnNext(func(i interface{}) { fmt.Printf("Second observer: %d\n", i) }) observable.Connect(context.Background()) time.Sleep(3 * time.Second)}运行输入：
$ go run main.gobefore subscribe second observerSecond observer: 1First observer: 1First observer: 2First observer: 3Second observer: 2Second observer: 3下面是等两个观察者都注册之后，并且手动调用了 Observable 的Connect()办法才产生数据。而且可连贯的 Observable 有一个个性：它是冷启动的！！！，即每个观察者都会收到一份雷同的拷贝。
转换 Observable
rxgo 提供了很多转换函数，能够批改通过它的rxgo.Item，而后再发送给下一个阶段。
Map
Map()办法简略批改它收到的rxgo.Item而后发送到下一个阶段（转换或过滤）。Map()承受一个类型为func (context.Context, interface{}) (interface{}, error)的函数。第二个参数就是rxgo.Item中的数据，返回转换后的数据。如果出错，则返回谬误。
func main() { observable := rxgo.Just(1, 2, 3)() observable = observable.Map(func(_ context.Context, i interface{}) (interface{}, error) { return i.(int)*2 + 1, nil }).Map(func(_ context.Context, i interface{}) (interface{}, error) { return i.(int)*3 + 2, nil }) for item := range observable.Observe() { fmt.Println(item.V) }}上例中每个数字通过两个Map，第一个Map执行2 * i + 1，第二个Map执行3 * i + 2。即对于每个数字来说，最终进行的变换为3 * (2 * i + 1) + 2。运行：
$ go run main.go111723Marshal
Marshal对通过它的数据进行一次Marshal。这个Marshal能够是json.Marshal/proto.Marshal，甚至咱们本人写的Marshal函数。它承受一个类型为func(interface{}) ([]byte, error)的函数用于对数据进行解决。
type User struct { Name string `json:"name"` Age int `json:"age"`}func main() { observable := rxgo.Just( User{ Name: "dj", Age: 18, }, User{ Name: "jw", Age: 20, }, )() observable = observable.Marshal(json.Marshal) for item := range observable.Observe() { fmt.Println(string(item.V.([]byte))) }}因为Marshal操作返回的是[]byte类型，咱们须要进行类型转换之后再输入。
Unmarshal
既然有Marshal，也就有它的相同操作Unmarshal。Unmarshal用于将一个[]byte类型转换为相应的构造体或其余类型。与Marshal不同，Unmarshal须要晓得转换的指标类型，所以须要提供一个函数用于生成该类型的对象。而后将[]byte数据Unmarshal到该对象中。Unmarshal承受两个参数，参数一是类型为func([]byte, interface{}) error的函数，参数二是func () interface{}用于生成理论类型的对象。咱们拿下面的例子中生成的 JSON 字符串作为数据，将它们从新Unmarshal为User对象：
type User struct { Name string `json:"name"` Age int `json:"age"`}func main() { observable := rxgo.Just( `{"name":"dj","age":18}`, `{"name":"jw","age":20}`, )() observable = observable.Map(func(_ context.Context, i interface{}) (interface{}, error) { return []byte(i.(string)), nil }).Unmarshal(json.Unmarshal, func() interface{} { return &User{} }) for item := range observable.Observe() { fmt.Println(item.V) }}因为Unmarshaller承受[]byte类型的参数，咱们在Unmarshal之前加了一个Map用于将string转为[]byte。运行：
$ go run main.go&{dj 18}&{jw 20}Buffer
Buffer依照肯定的规定收集接管到的数据，而后一次性发送进来（作为切片），而不是收到一个发送一个。有 3 种类型的Buffer：
BufferWithCount(n)：每收到n个数据发送一次，最初一次可能少于n个；BufferWithTime(n)：发送在一个工夫距离n内收到的数据；BufferWithTimeOrCount(d, n)：收到n个数据，或通过d工夫距离，发送以后收到的数据。
BufferWithCount：
func main() { observable := rxgo.Just(1, 2, 3, 4)() observable = observable.BufferWithCount(3) for item := range observable.Observe() { fmt.Println(item.V) }}运行：
$ go run main.go[1 2 3][4]留神，最初一组只有一个。
BufferWithTime：
func main() { ch := make(chan rxgo.Item, 1) go func() { i := 0 for range time.Tick(time.Second) { ch <- rxgo.Of(i) i++ } }() observable := rxgo.FromChannel(ch).BufferWithTime(rxgo.WithDuration(3 * time.Second)) for item := range observable.Observe() { fmt.Println(item.V) }}每 3s 发送一次：
$ go run main.go[0 1 2][3 4 5][6 7 8]...BufferWithTimeOrCount：
func main() { ch := make(chan rxgo.Item, 1) go func() { i := 0 for range time.Tick(time.Second) { ch <- rxgo.Of(i) i++ } }() observable := rxgo.FromChannel(ch).BufferWithTimeOrCount(rxgo.WithDuration(3*time.Second), 2) for item := range observable.Observe() { fmt.Println(item.V) }}下面 3s 能够收集 3 个数据，然而设置了收集 2 个就发送。所以，运行输入为：
$ go run main.go[0 1][2 3][4 5]...GroupBy
GroupBy依据传入一个 Hash 函数，为每个不同的后果别离创立新的 Observable。换句话说，GroupBy生成一个数据类型为 Observable 的 Observable。
func main() { count := 3 observable := rxgo.Range(0, 10).GroupBy(count, func(item rxgo.Item) int { return item.V.(int) % count }, rxgo.WithBufferedChannel(10)) for subObservable := range observable.Observe() { fmt.Println("New observable:") for item := range subObservable.V.(rxgo.Observable).Observe() { fmt.Printf("item: %v\n", item.V) } }}下面依据每个数模 3 的余数将整个流分为 3 组。运行：
$ go run main.go New observable:item: 0item: 3item: 6item: 9New observable:item: 1item: 4item: 7item: 10New observable:item: 2item: 5item: 8留神rxgo.WithBufferedChannel(10)的应用，因为咱们的数字是间断生成的，顺次为 0->1->2->...->9->10。而 Observable 默认是惰性的，即由Observe()驱动。内层的Observe()在返回一个 0 之后就期待下一个数，然而下一个数 1 不在此 Observable 中。所以会陷入死锁。应用rxgo.WithBufferedChannel(10)，设置它们之间的连贯 channel 缓冲区大小为 10，这样即便咱们未取出 channel 外面的数字，上游还是能发送数字进来。
并行操作
默认状况下，这些转换操作都是串行的，即只有一个 goroutine 负责执行转换函数。咱们也能够应用rxgo.WithPool(n)选项设置运行n个 goroutine，或者rxgo.WitCPUPool()选项设置运行与逻辑 CPU 数量相等的 goroutine。
func main() { observable := rxgo.Range(1, 100) observable = observable.Map(func(_ context.Context, i interface{}) (interface{}, error) { time.Sleep(time.Duration(rand.Int31())) return i.(int)*2 + 1, nil }, rxgo.WithCPUPool()) for item := range observable.Observe() { fmt.Println(item.V) }}因为是并行，所以输入程序就不确定了。为了让不确定性更显著一点，我在代码中加了一行time.Sleep。
过滤 Observable
Observable 中发送过去的数据并不一定都是咱们须要的，咱们要把不想要的过滤掉。
Filter
Filter()承受一个类型为func (i interface{}) bool的参数，通过的数据应用这个函数断言，返回true的将发送给下一个阶段。否则，抛弃。
func main() { observable := rxgo.Range(1, 10) observable = observable.Filter(func(i interface{}) bool { return i.(int)%2 == 0 }) for item := range observable.Observe() { fmt.Println(item.V) }}下面过滤掉奇数，最初只剩下偶数：
$ go run main.go246810ElementAt
ElementAt()只发送指定索引的数据，如ElementAt(2)只发送索引为 2 的数据，即第 3 个数据。
func main() { observable := rxgo.Just(0, 1, 2, 3, 4)().ElementAt(2) for item := range observable.Observe() { fmt.Println(item.V) }}下面代码输入 2。
Debounce
Debounce()比拟有意思，它收到数据后还会期待指定的工夫距离，后续距离内没有收到其余数据才会发送刚开始的数据。
func main() { ch := make(chan rxgo.Item) go func() { ch <- rxgo.Of(1) time.Sleep(2 * time.Second) ch <- rxgo.Of(2) ch <- rxgo.Of(3) time.Sleep(2 * time.Second) close(ch) }() observable := rxgo.FromChannel(ch).Debounce(rxgo.WithDuration(1 * time.Second)) for item := range observable.Observe() { fmt.Println(item.V) }}下面示例，先收到 1，而后 2s 内没收到数据，所以发送 1。接着收到了数据 2，因为马上又收到了 3，所以 2 不会发送。收到 3 之后 2s 内没有收到数据，发送了 3。所以最初输入为 1，3。
Distinct
Distinct()会记录它发送的所有数据，它不会发送反复的数据。因为数据格式多样，Distinct()要求咱们提供一个函数，依据原数据返回一个惟一标识码（有点相似哈希值）。基于这个标识码去重。
func main() { observable := rxgo.Just(1, 2, 2, 3, 3, 4, 4)(). Distinct(func(_ context.Context, i interface{}) (interface{}, error) { return i, nil }) for item := range observable.Observe() { fmt.Println(item.V) }}顺次输入 1，2，3，4，没有反复。
Skip
Skip能够跳过前若干个数据。
func main() { observable := rxgo.Just(1, 2, 3, 4, 5)().Skip(2) for item := range observable.Observe() { fmt.Println(item.V) }}Take
Take只取前若干个数据。
func main() { observable := rxgo.Just(1, 2, 3, 4, 5)().Take(2) for item := range observable.Observe() { fmt.Println(item.V) }}选项
rxgo 提供的大部分办法的最初一个参数是一个可变长的选项类型。这是 Go 中特有的、经典的选项设计模式。咱们后面曾经应用了：
rxgo.WithBufferedChannel(10)：设置 channel 的缓存大小；rxgo.WithPool(n)/rxgo.WithCpuPool()：应用多个 goroutine 执行转换操作；rxgo.WithPublishStrategy()：应用公布策略，即创立可连贯的 Observable。
除此之外，rxgo 还提供了很多其余选项。留待大家自行摸索了。
总结
rxgo 让基于 pipelines 的并发编程变得更容易！
大家如果发现好玩、好用的 Go 语言库，欢送到 Go 每日一库 GitHub 上提交 issue????
参考
- rxgo GitHub：https://github.com/jordan-wright/rxgo
- Go 每日一库 GitHub：https://github.com/darjun/go-daily-lib
我
我的博客：https://darjun.github.io
欢送关注我的微信公众号【GoUpUp】，独特学习，一起提高~