最近的項目用到了 RxGo ，因為之前從沒有接觸過，特意去學了學，特此記錄下。文章很多內容是復制了參考資料或者官方文檔。如果涉及侵權，請聯(lián)系刪除，謝謝。

1、RxGo簡介

1.1 基礎介紹

RxGo是一個基于Go語言的響應式編程庫，它提供了一種簡單而強大的方式來處理異步事件流和數(shù)據(jù)流。RxGo的設計靈感來自于ReactiveX，它提供了類似于ReactiveX的操作符和概念，如Observable、Observer、Subject、Scheduler等。

RxGo的目標是提供一種簡單而強大的方式來處理異步事件流和數(shù)據(jù)流，使得開發(fā)人員可以更容易地編寫高效、可維護和可擴展的代碼。RxGo的特點包括：

(資料圖片僅供參考)

1. 響應式編程：RxGo提供了Observable和Observer兩個核心概念，使得開發(fā)人員可以更容易地處理異步事件流和數(shù)據(jù)流。
2. 操作符：RxGo提供了類似于ReactiveX的操作符，如map、filter、reduce等，使得開發(fā)人員可以更容易地對事件流進行轉換、過濾和聚合等操作。
3. 調度器：RxGo提供了調度器，使得開發(fā)人員可以更容易地控制事件流的執(zhí)行線程和順序。
4. 可組合性：RxGo的操作符具有可組合性，使得開發(fā)人員可以更容易地組合多個操作符來實現(xiàn)復雜的操作。
5. 高效性：RxGo的設計和實現(xiàn)都非常高效，可以處理大量的事件流和數(shù)據(jù)流。

總之，RxGo是一個非常強大和實用的響應式編程庫，它可以幫助開發(fā)人員更容易地處理異步事件流和數(shù)據(jù)流，提高代碼的可維護性和可擴展性。

1.2 RxGo 數(shù)據(jù)流程圖

RxGo的實現(xiàn)基于管道的概念。管道是由通道連接的一系列階段，其中每個階段是運行相同功能的一組goroutine。

• 使用Just操作符創(chuàng)建一個基于固定列表的靜態(tài)可觀測數(shù)據(jù)。
• 使用Map操作符定義了一個轉換函數(shù)（把圓形變成方形）。
• 用Filter操作符過濾掉黃色方形。

從上面的例子中可以看出來，最終生成的數(shù)據(jù)被發(fā)送到一個通道中，消費者讀取數(shù)據(jù)進行消費。RxGo中有很多種消費和生成數(shù)據(jù)的方式，發(fā)布結果到通道中只是其中一種方式。

2、快速入門

2.1 安裝 RxGo v2

go get -u github.com/reactivex/rxgo/v2

2.2 簡單案例

我們先寫一個簡單的案例，來學習RxGo的簡單使用。

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創(chuàng)建 Observable，Operator就是用來創(chuàng)建 Observable的。
• 中間各個階段可以使用過濾操作篩選出我們想要的數(shù)據(jù)，使用轉換操作對數(shù)據(jù)進行轉換；
• 調用 Observable的Observe()方法，該方法返回一個<- chan rxgo.Item。然后for range遍歷即可。

結合上面的這張圖，我們就比較容易理解RxGo的數(shù)據(jù)處理流程。因為例子比較簡單，沒有用到Map、Filter操作。

執(zhí)行結果：

$ go run main.go 12345

Just使用到柯里化的編程思想。柯里化（Currying）是一種函數(shù)式編程的技術，它將一個接受多個參數(shù)的函數(shù)轉換成一系列接受單個參數(shù)的函數(shù)。這些單參數(shù)函數(shù)可以被組合起來，以便在后續(xù)的計算中使用。

柯里化的主要優(yōu)點是它可以使函數(shù)更加靈活和可復用。通過將函數(shù)分解為一系列單參數(shù)函數(shù)，我們可以更容易地組合和重用這些函數(shù)，從而減少代碼的重復性和冗余性。

例如：

//柯里化的例子func addCurried(x int) func(int) int {return func(y int) int {return x + y}}func main()  {add5 := addCurried(5)fmt.Println(add5(10))}

由于 Go 不支持多個可變參數(shù)，Just通過柯里化迂回地實現(xiàn)了這個功能：

//Just creates an Observable with the provided items.func Just(items ...interface{}) func(opts ...Option) Observable {  return func(opts ...Option) Observable {    return &ObservableImpl{      iterable: newJustIterable(items...)(opts...),    }  }}

Observe()返回一個 Item 的chan ，Item的結構如下：

// Item is a wrapper having either a value or an error.typeItem struct {V interface{}E error}

所以通過Just生成observable對象時，傳入的數(shù)據(jù)可以包含錯誤，在使用時通過 item.Error() 來區(qū)分。

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)    }  }}

我們使用item.Error()檢查是否出現(xiàn)錯誤。然后使用item.V訪問數(shù)據(jù)，item.E訪問錯誤。

除了使用for range之外，我們還可以調用 Observable的ForEach()方法來實現(xiàn)遍歷。ForEach()接受 3 個回調函數(shù)：

• NextFunc：類型為func (v interface {})，傳入的數(shù)據(jù)不包含錯誤類型時走此函數(shù)處理。
• ErrFunc：類型為func (err error)，當傳入的數(shù)據(jù)包含錯誤時走此函數(shù)；
• CompletedFunc：類型為func ()，Observable完成時調用。

有點Promise那味了。使用ForEach()，可以將上面的示例改寫為：

func main() {  observable := rxgo.Just(1, 2, errors.New("這是一個測試錯誤！"), 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: 2error: 這是一個測試錯誤！received: 4received: 5completed

ForEach()返回的是一個 chan，用于當 observable 關閉時會向此chan發(fā)送數(shù)據(jù)。所以在 observable前面加了<-來阻塞等待 ForEach()處理完數(shù)據(jù)。

3、RxGo 深入學習

上面的簡單案例，我們是使用Just來創(chuàng)建observable。其實還有其他的方式創(chuàng)建observable。一起來看一看。

3.1 rxgo.Create

傳入一個[]rxgo.Producer的切片，其中rxgo.Producer的類型為func(ctx context.Context, next chan<- Item)。我們可以在代碼中調用rxgo.Of(value)生成數(shù)據(jù)，rxgo.Error(err)生成錯誤，然后發(fā)送到next通道中：

package mainimport ("context""errors""fmt""github.com/reactivex/rxgo/v2")func main()  {observable := rxgo.Create([]rxgo.Producer{func(ctx context.Context, next chan<- rxgo.Item) {next <- rxgo.Of(1)next <- rxgo.Of("aaa")next <- rxgo.Of(errors.New("test"))}})ch := observable.Observe()for item := range ch {if item.Error() {fmt.Println("err:", item.E)}else {fmt.Println(item.V)}}}

因為rxgo.Create中的參數(shù)是[]rxgo.Producer，所以分成兩個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)}})

3.2 rxgo.FromChannel

FromChannel可以直接從一個已存在的<-chan rxgo.Item對象中創(chuàng)建 Observable：

package mainimport ("fmt""github.com/reactivex/rxgo/v2")func main()  {ch := make(chan rxgo.Item)go func() {for i := 0; i < 5; i++ {ch <- rxgo.Of(i)}//需要手動關閉 ch 通道close(ch)}()observable := rxgo.FromChannel(ch)for item := range observable.Observe() {if item.Error() {fmt.Println("err:", item.E)}else {fmt.Println(item.V)}}}

注意:

通道需要手動調用close()關閉，上面Create()方法內部rxgo自動幫我們執(zhí)行了這個步驟。

func newCreateIterable(fs []Producer, opts ...Option) Iterable {...go func() {// Create方法內部自動關閉了 next 通道defer close(next)for _, f := range fs {f(ctx, next)}}()...}

3.3 rxgo.Interval

Interval以傳入的時間間隔生成一個無窮的數(shù)字序列，從 0 開始：

func main()  {observable := rxgo.Interval(rxgo.WithDuration(time.Second))for item := range observable.Observe() {if item.Error() {fmt.Println("err:", item.E)}else {fmt.Println(item.V)}}}

運行后，第一秒輸出 0，第二秒輸出 1，以此類推。

3.4 rxgo.Range

func main() {  observable := rxgo.Range(0, 3)  for item := range observable.Observe() {    fmt.Println(item.V)  }}

Range可以生成一個范圍內的數(shù)字：

上面代碼依次輸出 0，1，2，3。

3.5 Repeat

這個和之前的不太一樣，這個是對已經存在的 observable對象調用 Repeat方法，從而實現(xiàn)重復生成數(shù)據(jù)。

package mainimport ("fmt""github.com/reactivex/rxgo/v2""time")func main()  {observable := rxgo.Range(0,3).Repeat(2, rxgo.WithDuration(time.Second))for item := range observable.Observe() {if item.Error() {fmt.Println("err:", item.E)}else {fmt.Println(item.V)}}}

輸出：

012012012

注意：這里執(zhí)行的次數(shù)一共是3次，Repeat中的參數(shù)是2，重復2次，一共3次。

3.6 rxgo.Start

可以給Start方法傳入[]rxgo.Supplier作為參數(shù)，它可以包含任意數(shù)量的rxgo.Supplier類型。rxgo.Supplier的底層類型為：

var Supplier func(ctx context.Context) rxgo.Item

Observable內部會依次調用這些rxgo.Supplier生成rxgo.Item：

package mainimport ("context""fmt""github.com/reactivex/rxgo/v2""time")func Supplier1(ctx context.Context) rxgo.Item {deadline, ok  := ctx.Deadline()fmt.Println("Supplier1", deadline, ok)time.Sleep(time.Second)return rxgo.Of(1)}func Supplier2(ctx context.Context) rxgo.Item {deadline, ok  := ctx.Deadline()fmt.Println("Supplier2", deadline, ok)time.Sleep(time.Second)return rxgo.Of(2)}func Supplier3(ctx context.Context) rxgo.Item {deadline, ok  := ctx.Deadline()fmt.Println("Supplier3", deadline, ok)time.Sleep(time.Second)return rxgo.Of(3)}func main() {ctx, _ := context.WithTimeout(context.Background(), time.Second*2)observable := rxgo.Start([]rxgo.Supplier{Supplier1, Supplier2, Supplier3}, rxgo.WithContext(ctx))for item := range observable.Observe() {fmt.Println(item.V)}}

4、Observable 分類

根據(jù)數(shù)據(jù)在何處生成，Observable被分為 Hot和 Cold兩種類型。

• Hot Observable：熱可觀測量，數(shù)據(jù)由可觀測量外部產生。
• Cold Observable：冷可觀測量，數(shù)據(jù)由可觀測量內部產生。

通常不想一次性的創(chuàng)建所有的數(shù)據(jù)，使用 熱可觀測量。

4.1 熱可觀測量示例

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)  }}

結果：

012

上面創(chuàng)建的是 Hot Observable。但是有個問題，第一次Observe()消耗了所有的數(shù)據(jù)，第二個就沒有數(shù)據(jù)輸出了。(可以用可連接的觀測量來修改這一行為，后面再說)。

4.2 冷可觀測量示例

Cold Observable就不會有這個問題，因為它創(chuàng)建的流是獨立于每個觀察者的。即每次調用Observe()都創(chuàng)建一個新的 channel。我們使用Defer()方法創(chuàng)建 Cold Observable，它的參數(shù)與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)  }}

Defer源碼介紹：

// Defer does not create the Observable until the observer subscribes,// and creates a fresh Observable for each observer.func Defer(f []Producer, opts ...Option) Observable {return &ObservableImpl{iterable: newDeferIterable(f, opts...),}}

執(zhí)行結果：

$ go run main.go012012

4.3 可連接的 Observable

可連接的（Connectable）Observable對普通的 Observable進行了一層組裝。調用它的Observe()方法時并不會立刻產生數(shù)據(jù)。使用它，我們可以等所有的觀察者都準備就緒了（即調用了Observe()方法）之后，再調用其Connect()方法開始生成數(shù)據(jù)。我們通過兩個示例比較使用普通的 Observable和可連接的 Observable有何不同。

4.3.1 普通的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()方法是異步執(zhí)行的，所以為了等待結果輸出，在最后增加了一行time.Sleep。運行結果：

First observer: 1First observer: 2First observer: 3before subscribe second observer

由輸出可以看出，注冊第一個觀察者之后就開始產生數(shù)據(jù)了。第二個觀察者并不會得到數(shù)據(jù)。

4.3.2 可連接的Observable

通過在創(chuàng)建 Observable的方法中指定rxgo.WithPublishStrategy()選項就可以創(chuàng)建可連接的 Observable：

• 重點是傳入rxgo.WithPublishStrategy()

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 才會產生數(shù)據(jù)  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()方法才產生數(shù)據(jù)。而且可連接的 Observable有一個特性：它是冷啟動的！！！，即每個觀察者都會收到一份相同的拷貝。

5、轉換 Observable

通過 RxGo 數(shù)據(jù)流程圖我們知道，我們可以對rxgo.Item進行轉換。rxgo 提供了很多轉換函數(shù)，下面一起來學一學這些轉換函數(shù)。

5.1 Map

Map()方法簡單修改它收到的rxgo.Item然后發(fā)送到下一個階段（轉換或過濾）。Map()接受一個類型為func (context.Context, interface{}) (interface{}, error)的函數(shù)。第二個參數(shù)就是rxgo.Item中的數(shù)據(jù)，返回轉換后的數(shù)據(jù)。如果出錯，則返回錯誤。

func main() {observable := rxgo.Just(1, 2, 3)()observable = observable.Map(func(_ context.Context, i interface{}) (interface{}, error) {return i.(int), nil}).Map(func(_ context.Context, i interface{}) (interface{}, error) {b := i.(int)if b % 2 == 0 {return nil, errors.New("test")} else {return i, nil}})for item := range observable.Observe() {fmt.Println(item.V)}}

上例中每個數(shù)字經過兩個Map，第一個Map邏輯是原樣輸出，第二個Map邏輯是判斷i是不是偶數(shù)，如果是偶數(shù)，就返回錯誤，否則原樣輸出。運行結果：

1

我們將第一個Map中的語句改為下面的邏輯：

return i.(int) + 1, nil

運行結果：

我們可以知道，數(shù)據(jù)的處理是串行的，第一個數(shù)據(jù)執(zhí)行完所有的Map過后，第二個數(shù)據(jù)才會執(zhí)行，當其中某一個執(zhí)行返回的結果包含錯誤，就不會繼續(xù)進行轉換了，即不會數(shù)據(jù)不會進入到 Observe() 中的通道中去。

5.2 Marshal

Marshal對經過它的數(shù)據(jù)進行一次Marshal。這個Marshal可以是json.Marshal/proto.Marshal，甚至我們自己寫的Marshal函數(shù)。它接受一個類型為func(interface{}) ([]byte, error)的函數(shù)用于對數(shù)據(jù)進行處理。

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)))  }}

執(zhí)行結果：

{"name":"dj","age":18}{"name":"jw","age":20}

由于Marshal操作返回的是[]byte類型，我們需要進行類型轉換之后再輸出。

5.3 Unmarshal

既然有Marshal，也就有它的相反操作Unmarshal。Unmarshal用于將一個[]byte類型轉換為相應的結構體或其他類型。與Marshal不同，Unmarshal需要知道轉換的目標類型，所以需要提供一個函數(shù)用于生成該類型的對象。然后將[]byte數(shù)據(jù)Unmarshal到該對象中。Unmarshal接受兩個參數(shù)，參數(shù)一是類型為func([]byte, interface{}) error的函數(shù)，參數(shù)二是func () interface{}用于生成實際類型的對象。我們拿上面的例子中生成的 JSON 字符串作為數(shù)據(jù)，將它們重新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類型的參數(shù)，我們在Unmarshal之前加了一個Map用于將string轉為[]byte。運行結果：

&{dj 18}&{jw 20}

5.4 Buffer

Buffer按照一定的規(guī)則收集接收到的數(shù)據(jù)，然后一次性發(fā)送出去（作為切片），而不是收到一個發(fā)送一個。有 3 種類型的Buffer：

• BufferWithCount(n)：每收到n個數(shù)據(jù)發(fā)送一次，最后一次可能少于n個；
• BufferWithTime(n)：發(fā)送在一個時間間隔n內收到的數(shù)據(jù)；
• BufferWithTimeOrCount(d, n)：收到n個數(shù)據(jù)，或經過d時間間隔，發(fā)送當前收到的數(shù)據(jù)。

5.4.1 BufferWithCount

func main() {observable := rxgo.Range(0, 5)observable = observable.BufferWithCount(2)for item := range observable.Observe() {fmt.Println(item.V)}}

執(zhí)行結果：

[0 1][2 3][4]

最后一組只有一個。

5.4.2 BufferWithTime

unc main() {ch := make(chan rxgo.Item, 1)go func() {i := 0for range time.Tick(time.Second) {ch <- rxgo.Of(i)i++}}()observable := rxgo.FromChannel(ch).BufferWithTime(rxgo.WithDuration(2 * time.Second))layout := "2006-01-02 13:04:05"fmt.Println("startTime", time.Now().Format(layout))for item := range observable.Observe() {fmt.Println(item.V)fmt.Println("nextTime", time.Now().Format(layout))}}

執(zhí)行結果是不確定的，這里需要注意：

startTime 2023-04-22 44:15:49[0]nextTime 2023-04-22 44:15:51[1 2]nextTime 2023-04-22 44:15:53[3 4 5]nextTime 2023-04-22 44:15:55...

5.4.3 BufferWithTimeOrCount

func main() {ch := make(chan rxgo.Item, 1)go func() {i := 0for range time.Tick(time.Second) {ch <- rxgo.Of(i)i++}}()observable := rxgo.FromChannel(ch).BufferWithTimeOrCount(rxgo.WithDuration(2*time.Second), 2)layout := "2006-01-02 13:04:05"fmt.Println("startTime", time.Now().Format(layout))for item := range observable.Observe() {fmt.Println(item.V)fmt.Println("nextTime", time.Now().Format(layout))}}

執(zhí)行結果：

startTime 2023-04-22 44:18:48[0]nextTime 2023-04-22 44:18:50[1 2]nextTime 2023-04-22 44:18:51[3 4]nextTime 2023-04-22 44:18:53

BufferWithTimeOrCount是以BufferWithCount、BufferWithTime誰先滿足條件為準，誰先滿足誰就先執(zhí)行。

5.5 GroupBy

``GroupBy將一個Observable分成多個子Observable，每個子Observable`包含相同的索引值的元素。

GroupBy函數(shù)定義如下：

GroupBy(length int, distribution func(Item) int, opts ...Option) Observable

即將一個Observable分成length個子Observable，根據(jù)distribution函數(shù)返回的int作為分組的依據(jù)。

package mainimport ("fmt""github.com/reactivex/rxgo/v2")func main() {// 創(chuàng)建一個Observable，它發(fā)出一些整數(shù)值source := rxgo.Just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)()// 使用GroupBy操作符將整數(shù)值按照奇偶性進行分組grouped := source.GroupBy(2, func(item rxgo.Item) int {return item.V.(int) % 2}, rxgo.WithBufferedChannel(10))for subObservable := range grouped.Observe() {fmt.Println("new subObservable ------ ")for item := range subObservable.V.(rxgo.Observable).Observe() {fmt.Printf("%v\n", item.V)}}}

上面根據(jù)每個數(shù)模 3 的余數(shù)將整個流分為 3 組。運行：

new subObservable ------ 246810new subObservable ------ 13579

注意rxgo.WithBufferedChannel(10)的使用，由于我們的數(shù)字是連續(xù)生成的，依次為 0->1->2->…->9->10。而 Observable默認是惰性的，即由Observe()驅動。內層的Observe()在返回一個 0 之后就等待下一個數(shù)，但是下一個數(shù) 1 不在此 Observable中。所以會陷入死鎖。使用rxgo.WithBufferedChannel(10)，設置它們之間的連接 channel 緩沖區(qū)大小為 10，這樣即使我們未取出 channel 里面的數(shù)字，上游還是能發(fā)送數(shù)字進來。

6、并行操作

默認情況下，這些轉換操作都是串行的，即只有一個 goroutine 負責執(zhí)行轉換函數(shù)。從上面的Map操作也可以得知默認是串行執(zhí)行的。可以改變這一默認行為，使用rxgo.WithPool(n)選項設置運行n個 goroutine，或者rxgo.WitCPUPool()選項設置運行與邏輯 CPU 數(shù)量相等的 goroutine。

package mainimport ("context""fmt""github.com/reactivex/rxgo/v2""math/rand""time")func main() {observable := rxgo.Range(1, 10)observable = observable.Map(func(_ context.Context, i interface{}) (interface{}, error) {time.Sleep(time.Duration(rand.Int31()))return i.(int) + 1, nil}, rxgo.WithCPUPool())for item := range observable.Observe() {fmt.Println(item.V)}}

結果：

891065112473

由于是并行運算，所以結果是不固定的。

我們可以直接看官網的介紹：https://github.com/ReactiveX/RxGo/blob/v2.5.0/doc/options.md

7、過濾 Observable

我們可以對Observable中發(fā)送過來的數(shù)據(jù)進行過濾，過濾掉不需要的數(shù)據(jù)，有以下方式：

• Filter

• ElementAt

• Debounce

• Distinct

• Skip

• Take

下面的內容大多來自官方的示例，地址：https://github.com/ReactiveX/RxGo/tree/v2.5.0/doc

7.1 Filter

Filter()接受一個類型為func (i interface{}) bool的參數(shù)，通過的數(shù)據(jù)使用這個函數(shù)斷言，返回true的將發(fā)送給下一個階段。否則，丟棄。

package mainimport ("fmt""github.com/reactivex/rxgo/v2")func main() {observable := rxgo.Just(1, 2, 3)().Filter(func(i interface{}) bool {return i != 2})for item := range observable.Observe() {fmt.Println(item.V)}}

結果：

13

7.2 ElementAt

ElementAt()只發(fā)送指定索引的數(shù)據(jù)，如ElementAt(2)只發(fā)送索引為 2 的數(shù)據(jù)，即第 3 個數(shù)據(jù)。

package mainimport ("fmt""github.com/reactivex/rxgo/v2")func main() {observable := rxgo.Just(0, 1, 2, 3, 4)().ElementAt(2)for item := range observable.Observe() {fmt.Println(item.V)}}

結果：

2

7.3 Debounce

只有當特定的時間跨度已經過去而沒有發(fā)出另一個Item時，才從Observable發(fā)出一個Item。

package mainimport ("fmt""github.com/reactivex/rxgo/v2""time")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)}}

結果：

13

上面示例，先收到 1，然后 2s 內沒收到數(shù)據(jù)，所以發(fā)送 1。接著收到了數(shù)據(jù) 2，由于馬上又收到了 3，所以 2 不會發(fā)送。收到 3 之后 2s 內沒有收到數(shù)據(jù)，發(fā)送了 3。所以最后輸出為 1，3。

7.4 Distinct

Distinct()會記錄它發(fā)送的所有數(shù)據(jù)，它不會發(fā)送重復的數(shù)據(jù)。由于數(shù)據(jù)格式多樣，Distinct()要求我們提供一個函數(shù)，根據(jù)原數(shù)據(jù)返回一個唯一標識碼（有點類似哈希值）。基于這個標識碼去重。

package mainimport ("context""fmt""github.com/reactivex/rxgo/v2")func main() {observable := rxgo.Just(1, 2, 2, 3, 4, 4, 5)().Distinct(func(_ context.Context, i interface{}) (interface{}, error) {return i, nil})for item := range observable.Observe() {fmt.Println(item.V)}}

結果：

12345

7.5 Skip

Skip可以跳過前若干個數(shù)據(jù)。

package mainimport ("fmt""github.com/reactivex/rxgo/v2")func main() {observable := rxgo.Just(1, 2, 3, 4, 5)().Skip(2)for item := range observable.Observe() {fmt.Println(item.V)}}

結果：

345

7.6 Take

Take只取前若干個數(shù)據(jù)。

package mainimport ("fmt""github.com/reactivex/rxgo/v2")func main() {observable := rxgo.Just(1, 2, 3, 4, 5)().Take(2)for item := range observable.Observe() {fmt.Println(item.V)}}

結果：

12

8、選項

因為golang中不支持默認參數(shù)，所以我們經常會用到選項設計模式，rxgo中也大量使用到了此模式。

• rxgo.WithBufferedChannel(10)：設置 channel 的緩存大小；
• rxgo.WithPool(n)/rxgo.WithCpuPool()：使用多個 goroutine 執(zhí)行轉換操作；
• rxgo.WithPublishStrategy()：使用發(fā)布策略，即創(chuàng)建可連接的 Observable。

rxgo還有很多其他選項，具體看官方文檔，地址：

https://github.com/ReactiveX/RxGo/blob/v2.5.0/doc/options.md

9、簡化的真實案例

假設現(xiàn)在有一個定時處理任務，結構如下：

type ScheduledTask struct {RecordId intHandleStartTime time.TimeStatus bool}

在執(zhí)行具體的任務時，需要去數(shù)據(jù)庫查詢下是否已經被取消了，如果已經被取消掉的，則不再執(zhí)行。

完整代碼如下：

package mainimport ("fmt""github.com/reactivex/rxgo/v2""time")type ScheduledTask struct {RecordId intHandleStartTime stringStatus bool}func main() {ch := make(chan rxgo.Item)go producer(ch)time.Sleep(time.Second*3)observable := rxgo.FromChannel(ch)observable = observable.Filter(func(i interface{}) bool {st := i.(*ScheduledTask)return st.Status}, rxgo.WithBufferedChannel(1))// 消費可觀測量for customer := range observable.Observe() {st := customer.V.(*ScheduledTask)fmt.Printf("resutl: --> %+v\n", st)}}func producer(ch chan <- rxgo.Item)  {for i := 0; i < 10; i++ {status := falseif i % 2 == 0 {status = true}st := &ScheduledTask{RecordId: i,HandleStartTime: time.Now().Format("2006-01-02 13:04:05"),Status: status,}ch <- rxgo.Of(st)}  // 這里千萬不要忘記了close(ch)}

結果：

resutl: --> &{RecordId:0 HandleStartTime:2023-04-22 46:04:07 Status:true}resutl: --> &{RecordId:2 HandleStartTime:2023-04-22 46:04:10 Status:true}resutl: --> &{RecordId:4 HandleStartTime:2023-04-22 46:04:10 Status:true}resutl: --> &{RecordId:6 HandleStartTime:2023-04-22 46:04:10 Status:true}resutl: --> &{RecordId:8 HandleStartTime:2023-04-22 46:04:10 Status:true}

參考鏈接

Go 每日一庫之 rxgo

[官方例子](

關鍵詞：