如何在 RxGo 中玩乐

Question

我正在尝试使用 DoOnNext 实现反应流程的最后一步，以实现最后一步的并行执行。

运行下面的代码，我希望 thirdCounter = 2 以及每个“第一个 DoOnNext”、“第二个 DoOnNext”和 “第三个 DoOnNext” 将被打印两次（共 6 次）

打印符合预期，地图也正确连接了字符串。但是，thirdCounter = 7 因此步骤被过度调用。

我在这里缺少什么？

我的代码：

var thirdCounter int32
func localRun(names ...string) {
    observable := rxgo.Just(names)().
        Map(func(_ context.Context, i interface{}) (interface{}, error) {
            s := i.(string)
            s = fmt.Sprintf("%s,%s", s, "one")
            return s, nil
        }).
        Map(func(_ context.Context, i interface{}) (interface{}, error) {
            s := i.(string)
            s = fmt.Sprintf("%s,%s", s, "two")
            return s, nil
        }).
        Map(func(_ context.Context, i interface{}) (interface{}, error) {
            atomic.AddInt32(&thirdCounter, 1)
            s := i.(string)
            s = fmt.Sprintf("%s,%s", s, "three")
            return s, nil
        })

    observable.DoOnNext(func(i interface{}) {
        fmt.Println("first DoOnNext", i)
    })

    observable.DoOnNext(func(i interface{}) {
        fmt.Println("second DoOnNext", i)
    })

    observable.DoOnNext(func(i interface{}) {
        fmt.Println("third DoOnNext", i)
    })

    for item := range observable.Last().Observe() {
        fmt.Println(item.V)
    }
    fmt.Printf("Third Counter = %d\n", thirdCounter)
}
func TestMocktFlow(t *testing.T) {
    cs := make([]string, 0)
    cs = append(cs, "Hello")
    cs = append(cs, "Hi")
    localRun(cs...)
}

Answer 1

不是 3 个订阅者。您的代码有 4 个订阅者。

1. 第一个 DoOnNext()
2. 第二个 DoOnNext()
3. 第三个 DoOnNext()
4. observable.Last()

所以你应该期望thirdCounter=8。 如果您等到所有订阅结束， 可以看到输出8。

将代码的最后一部分改成如下：

time.Sleep(time.Second) // added
fmt.Printf("Third Counter = %d\n", thirdCounter) // 8