Go中响应式编程库RxGo详细介绍

最近的项目用到了 RxGo ,因为之前从没有接触过,特意去学了学,特此记录下。文章很多内容是复制了参考资料或者官方文档。如果涉及侵权,请联系删除,谢谢。

1、RxGo简介

1.1 基础介绍

RxGo是一个基于Go语言的响应式编程库,它提供了一种简单而强大的方式来处理异步事件流和数据流。RxGo的设计灵感来自于ReactiveX,它提供了类似于ReactiveX的操作符和概念,如Observable、Observer、Subject、Scheduler等。

RxGo的目标是提供一种简单而强大的方式来处理异步事件流和数据流,使得开发人员可以更容易地编写高效、可维护和可扩展的代码。RxGo的特点包括:

  1. 响应式编程:RxGo提供了Observable和Observer两个核心概念,使得开发人员可以更容易地处理异步事件流和数据流。
  2. 操作符:RxGo提供了类似于ReactiveX的操作符,如map、filter、reduce等,使得开发人员可以更容易地对事件流进行转换、过滤和聚合等操作。
  3. 调度器:RxGo提供了调度器,使得开发人员可以更容易地控制事件流的执行线程和顺序。
  4. 可组合性:RxGo的操作符具有可组合性,使得开发人员可以更容易地组合多个操作符来实现复杂的操作。
  5. 高效性:RxGo的设计和实现都非常高效,可以处理大量的事件流和数据流。

总之,RxGo是一个非常强大和实用的响应式编程库,它可以帮助开发人员更容易地处理异步事件流和数据流,提高代码的可维护性和可扩展性。

1.2 RxGo 数据流程图

RxGo的实现基于管道的概念。管道是由通道连接的一系列阶段,其中每个阶段是运行相同功能的一组goroutine。

  • 使用Just操作符创建一个基于固定列表的静态可观测数据。
  • 使用Map操作符定义了一个转换函数(把圆形变成方形)。
  • Filter操作符过滤掉黄色方形。

从上面的例子中可以看出来,最终生成的数据被发送到一个通道中,消费者读取数据进行消费。RxGo中有很多种消费和生成数据的方式,发布结果到通道中只是其中一种方式。

2、快速入门

2.1 安装 RxGo v2

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

2.2 简单案例

我们先写一个简单的案例,来学习RxGo的简单使用。

package main

import (
  "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 创建 ObservableOperator 就是用来创建 Observable 的。
  • 中间各个阶段可以使用过滤操作筛选出我们想要的数据,使用转换操作对数据进行转换;
  • 调用 ObservableObserve()方法,该方法返回一个<- chan rxgo.Item。然后for range遍历即可。

结合上面的这张图,我们就比较容易理解RxGo的数据处理流程。因为例子比较简单,没有用到Map、Filter操作。

执行结果:

$ go run main.go 
1
2
3
4
5

Just使用到柯里化的编程思想。
柯里化(Currying)是一种函数式编程的技术,它将一个接受多个参数的函数转换成一系列接受单个参数的函数。这些单参数函数可以被组合起来,以便在后续的计算中使用。

柯里化的主要优点是它可以使函数更加灵活和可复用。通过将函数分解为一系列单参数函数,我们可以更容易地组合和重用这些函数,从而减少代码的重复性和冗余性。

例如:

//柯里化的例子
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 不支持多个可变参数,Just通过柯里化迂回地实现了这个功能:

//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.
type	Item struct {
		V interface{}
		E error
	}

所以通过Just生成observable对象时,传入的数据可以包含错误,在使用时通过 item.Error() 来区分。

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()检查是否出现错误。然后使用item.V访问数据,item.E访问错误。

除了使用for range之外,我们还可以调用 ObservableForEach()方法来实现遍历。ForEach()接受 3 个回调函数:

  • NextFunc:类型为func (v interface {}),传入的数据不包含错误类型时走此函数处理。
  • ErrFunc:类型为func (err error),当传入的数据包含错误时走此函数;
  • 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: 1
received: 2
error: 这是一个测试错误!
received: 4
received: 5
completed

ForEach()返回的是一个 chan,用于当 observable 关闭时会向此chan发送数据。所以在 observable前面加了 <-来阻塞等待 ForEach()处理完数据。

3、RxGo 深入学习

上面的简单案例,我们是使用Just来创建observable。其实还有其他的方式创建observable。一起来看一看。

3.1 rxgo.Create

传入一个[]rxgo.Producer的切片,其中rxgo.Producer的类型为func(ctx context.Context, next chan<- Item)。我们可以在代码中调用rxgo.Of(value)生成数据,rxgo.Error(err)生成错误,然后发送到next通道中:

package main

import (
	"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中的参数是[]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对象中创建 Observable

package main

import (
	"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自动帮我们执行了这个步骤。

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以传入的时间间隔生成一个无穷的数字序列,从 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可以生成一个范围内的数字:

上面代码依次输出 0,1,2,3。

3.5 Repeat

这个和之前的不太一样,这个是对已经存在的 observable对象调用 Repeat方法,从而实现重复生成数据。

package main

import (
	"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)
		}
	}
}

输出:

0
1
2
0
1
2
0
1
2

注意:这里执行的次数一共是3次,Repeat中的参数是2,重复2次,一共3次。

3.6 rxgo.Start

可以给Start方法传入[]rxgo.Supplier作为参数,它可以包含任意数量的rxgo.Supplier类型。rxgo.Supplier的底层类型为:

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

Observable 内部会依次调用这些rxgo.Supplier生成rxgo.Item

package main

import (
	"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 分类

根据数据在何处生成,Observable 被分为 HotCold 两种类型。

  • Hot Observable:热可观测量,数据由可观测量外部产生。
  • Cold Observable:冷可观测量,数据由可观测量内部产生。

通常不想一次性的创建所有的数据,使用 热可观测量。

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

结果:

0
1
2

上面创建的是 Hot Observable。但是有个问题,第一次Observe()消耗了所有的数据,第二个就没有数据输出了。(可以用可连接的观测量来修改这一行为,后面再说)。

4.2 冷可观测量示例

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

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

执行结果:

$ go run main.go
0
1
2
0
1
2

4.3 可连接的 Observable

可连接的(Connectable)Observable 对普通的 Observable 进行了一层组装。调用它的Observe()方法时并不会立刻产生数据。使用它,我们可以等所有的观察者都准备就绪了(即调用了Observe()方法)之后,再调用其Connect()方法开始生成数据。我们通过两个示例比较使用普通的 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()方法是异步执行的,所以为了等待结果输出,在最后增加了一行time.Sleep。运行结果:

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

由输出可以看出,注册第一个观察者之后就开始产生数据了。第二个观察者并不会得到数据。

4.3.2 可连接的Observable

通过在创建 Observable 的方法中指定rxgo.WithPublishStrategy()选项就可以创建可连接的 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 才会产生数据
  observable.Connect(context.Background())
  time.Sleep(3 * time.Second)
}

运行输出:

$ go run main.go
before subscribe second observer
Second observer: 1
First observer: 1
First observer: 2
First observer: 3
Second observer: 2
Second observer: 3

上面是等两个观察者都注册之后,并且手动调用了 Observable 的Connect()方法才产生数据。而且可连接的 Observable 有一个特性:它是冷启动的!!!,即每个观察者都会收到一份相同的拷贝。

5、转换 Observable

通过 RxGo 数据流程图我们知道,我们可以对rxgo.Item进行转换。rxgo 提供了很多转换函数,下面一起来学一学这些转换函数。

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

上例中每个数字经过两个Map,第一个Map逻辑是原样输出,第二个Map逻辑是判断i是不是偶数,如果是偶数,就返回错误,否则原样输出。运行结果:

1
<nil>

我们将第一个Map中的语句改为下面的逻辑:

return i.(int) + 1, nil

运行结果:

<nil>

我们可以知道,数据的处理是串行的,第一个数据执行完所有的Map过后,第二个数据才会执行,当其中某一个执行返回的结果包含错误,就不会继续进行转换了,即不会数据不会进入到 Observe() 中的通道中去。

5.2 Marshal

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

执行结果:

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

由于Marshal操作返回的是[]byte类型,我们需要进行类型转换之后再输出。

5.3 Unmarshal

既然有Marshal,也就有它的相反操作UnmarshalUnmarshal用于将一个[]byte类型转换为相应的结构体或其他类型。与Marshal不同,Unmarshal需要知道转换的目标类型,所以需要提供一个函数用于生成该类型的对象。然后将[]byte数据Unmarshal到该对象中。Unmarshal接受两个参数,参数一是类型为func([]byte, interface{}) error的函数,参数二是func () interface{}用于生成实际类型的对象。我们拿上面的例子中生成的 JSON 字符串作为数据,将它们重新UnmarshalUser对象:

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。运行结果:

&{dj 18}
&{jw 20}

5.4 Buffer

Buffer按照一定的规则收集接收到的数据,然后一次性发送出去(作为切片),而不是收到一个发送一个。有 3 种类型的Buffer

  • BufferWithCount(n):每收到n个数据发送一次,最后一次可能少于n个;
  • BufferWithTime(n):发送在一个时间间隔n内收到的数据;
  • BufferWithTimeOrCount(d, n):收到n个数据,或经过d时间间隔,发送当前收到的数据。
5.4.1 BufferWithCount
func main() {
	observable := rxgo.Range(0, 5)

	observable = observable.BufferWithCount(2)

	for item := range observable.Observe() {
		fmt.Println(item.V)
	}
}

执行结果:

[0 1]
[2 3]
[4]

最后一组只有一个。

5.4.2 BufferWithTime
unc 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(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))

	}
}

执行结果是不确定的,这里需要注意:

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 := 0
		for 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))
	}
}

执行结果:

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谁先满足条件为准,谁先满足谁就先执行。

5.5 GroupBy

``GroupBy将一个Observable分成多个子Observable,每个子Observable`包含相同的索引值的元素。

GroupBy函数定义如下:

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

即将一个Observable分成length个子Observable,根据distribution函数返回的int作为分组的依据。

package main

import (
	"fmt"

	"github.com/reactivex/rxgo/v2"
)

func main() {
	// 创建一个Observable,它发出一些整数值
	source := rxgo.Just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)()

	// 使用GroupBy操作符将整数值按照奇偶性进行分组
	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)
		}
	}

}

上面根据每个数模 3 的余数将整个流分为 3 组。运行:

new subObservable ------ 
2
4
6
8
10
new subObservable ------ 
1
3
5
7
9

注意rxgo.WithBufferedChannel(10)的使用,由于我们的数字是连续生成的,依次为 0->1->2->…->9->10。而 Observable 默认是惰性的,即由Observe()驱动。内层的Observe()在返回一个 0 之后就等待下一个数,但是下一个数 1 不在此 Observable 中。所以会陷入死锁。使用rxgo.WithBufferedChannel(10),设置它们之间的连接 channel 缓冲区大小为 10,这样即使我们未取出 channel 里面的数字,上游还是能发送数字进来。

6、并行操作

默认情况下,这些转换操作都是串行的,即只有一个 goroutine 负责执行转换函数。从上面的Map操作也可以得知默认是串行执行的。可以改变这一默认行为,使用rxgo.WithPool(n)选项设置运行n个 goroutine,或者rxgo.WitCPUPool()选项设置运行与逻辑 CPU 数量相等的 goroutine。

package main

import (
	"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)
	}
}

结果:

8
9
10
6
5
11
2
4
7
3

由于是并行运算,所以结果是不固定的。

我们可以直接看官网的介绍:https://github.com/ReactiveX/RxGo/blob/v2.5.0/doc/options.md

7、过滤 Observable

我们可以对Observable 中发送过来的数据进行过滤,过滤掉不需要的数据,有以下方式:

  • Filter

  • ElementAt

  • Debounce

  • Distinct

  • Skip

  • Take

下面的内容大多来自官方的示例,地址:https://github.com/ReactiveX/RxGo/tree/v2.5.0/doc

7.1 Filter

Filter()接受一个类型为func (i interface{}) bool的参数,通过的数据使用这个函数断言,返回true的将发送给下一个阶段。否则,丢弃。

package main

import (
	"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)
	}
}

结果:

1
3

7.2 ElementAt

ElementAt()只发送指定索引的数据,如ElementAt(2)只发送索引为 2 的数据,即第 3 个数据。

package main

import (
	"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

只有当特定的时间跨度已经过去而没有发出另一个Item时,才从Observable发出一个Item

package main

import (
	"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)
	}
}

结果:

1
3

上面示例,先收到 1,然后 2s 内没收到数据,所以发送 1。接着收到了数据 2,由于马上又收到了 3,所以 2 不会发送。收到 3 之后 2s 内没有收到数据,发送了 3。所以最后输出为 1,3。

7.4 Distinct

Distinct()会记录它发送的所有数据,它不会发送重复的数据。由于数据格式多样,Distinct()要求我们提供一个函数,根据原数据返回一个唯一标识码(有点类似哈希值)。基于这个标识码去重。

package main

import (
	"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)
	}
}

结果:

1
2
3
4
5

7.5 Skip

Skip可以跳过前若干个数据。

package main

import (
	"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)
	}
}

结果:

3
4
5

7.6 Take

Take只取前若干个数据。

package main

import (
	"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)
	}
}

结果:

1
2

8、选项

因为golang中不支持默认参数,所以我们经常会用到选项设计模式,rxgo中也大量使用到了此模式。

  • rxgo.WithBufferedChannel(10):设置 channel 的缓存大小;
  • rxgo.WithPool(n)/rxgo.WithCpuPool():使用多个 goroutine 执行转换操作;
  • rxgo.WithPublishStrategy():使用发布策略,即创建可连接的 Observable

rxgo还有很多其他选项,具体看官方文档,地址:

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

参考链接

Go 每日一库之 rxgo

[官方例子](

热门相关:首席的独宠新娘   第一神算:纨绔大小姐   网游之逆天飞扬   第一神算:纨绔大小姐   仗剑高歌