golang 函数式编程库samber/mo使用： Future

golang 函数式编程库samber/mo使用： Future

如果您对samber/mo库不了解， 请先阅读第一篇 Option

本节讲述Future的使用，它可以帮助我们处理异步编程问题。

示例

我们先来看看下面代码的示例， 注释解释了每一步的操作。

package main

import (
	"fmt"

	"github.com/samber/mo"
)

func main() {
	// resolve 在这里只是一个定义， NewFuture会以一个 goroutine 的方式执行 cb， 并且传递Future 的 resolve 和 reject
	value, err := mo.NewFuture(func(resolve func(string), reject func(error)) {
		// do something here
		if true { // 这里假定 do something 成功
			// 如果 do something 成功， 执行 resolve， 并传递一个值， 然后会执行 Then
			resolve("foobar")
		} else {
			// 告诉 do something 失败， 执行 reject， 并传递一个错误， 然后会执行 Catch
			reject(fmt.Errorf("failure"))
		}
	}).
		Then(func(s string) (string, error) {
			// 这里 s 就是 resolve 传递的值
			return s, nil
		}).
		Catch(func(err error) (string, error) {
			// 这里 err 就是 reject 传递的错误
			return "foobar", nil
		}).
		Finally(func(value string, err error) (string, error) {
			// 不管发生什么都会执行， value 是 resolve 传递的值， err 是 reject 传递的错误
			return value, nil
		}).
		Collect() // 等待 future 执行完毕， 并返回最终的值

	fmt.Println(value)
	fmt.Println(err)
	// Output:
	// foobar
	// 
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

源码解析

根据mo.NewFuture的实现， 可以看出该函数做的事情就是构造一个Future， 然后执行activate函数， activate实际就是用 goroutine 执行cb函数， 并且将 Future 的 resolve 和 reject函数作为参数传递给cb。

func NewFuture[T any](cb func(resolve func(T), reject func(error))) *Future[T] {
	future := Future[T]{
		cb:       cb,
		cancelCb: func() {},
		done:     make(chan struct{}),
	}

	future.active()

	return &future
}

func (f *Future[T]) active() {
	go f.cb(f.resolve, f.reject)
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

resolve的实现如下， 可以看到resolve做的事情就是用mo.OK包装value， 记录到result中，并且关闭f.done， 表明future已经完成。

resolve加锁的目的是为了确保后续Then或Finally不会同时进行。

func (f *Future[T]) resolve(value T) {
	f.mu.Lock()
	defer f.mu.Unlock()

	f.result = Ok(value)
	if f.next != nil { 
	    // 这里如果不为空，表明next先于something注册，需要执行
		f.next.activeSync()
	}
	close(f.done)
}
1
2
3
4
5
6
7
8
9
10
11

我们来看看Then的实现，这个函数先对f执行加锁， 然后构造一个新的Future，这个新的Future的cb函数就是为了判断f的执行结果， 如果f的result不是error， 就执行Then注册的回调 cb。 所以如果f.cb函数执行resolve后返回， f.result.IsError()为false， 会执行Then中的回调。

最后的select表示如果f已完成，用goroutine 执行Then中的回调。如果f还没有完成，则留待f.cb的resolve或reject执行Then的回调。两种情况都会直接返回f.next，不会阻塞。这样就实现了Future的串联。

func (f *Future[T]) Then(cb func(T) (T, error)) *Future[T] {
	f.mu.Lock()
	defer f.mu.Unlock()

	f.next = &Future[T]{
		cb: func(resolve func(T), reject func(error)) {
			if f.result.IsError() {
				reject(f.result.Error())
				return
			}
			newValue, err := cb(f.result.MustGet())
			if err != nil {
				reject(err)
				return
			}
			resolve(newValue)
		},
		cancelCb: func() {
			f.Cancel()
		},
		done: make(chan struct{}),
	}

	select {
	case <-f.done:
		f.next.active()
	default:
	}
	return f.next
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

Catch和Then的区别是如果f.result是error， 就执行Catch中的回调。Finally是不管f.result是什么， 都会执行Finally中的回调。

最后的Collect是用于等待Future执行完毕， 并返回最终的值。

func (f *Future[T]) Collect() (T, error) {
	<-f.done
	return f.result.Get()
}
1
2
3
4

还有 Result 和 Either方法， 用于获取Future的执行结果， 会阻塞直到Future执行完毕（也就是先执行Collect)