Go实现线程安全的缓存

2023-07-09 14:09•Android•阅读 4385

场景

某些函数调用频繁，但其计算却非常耗时，为了避免每次调用时都重新计算一遍，我们需要保存函数的计算结果，这样在对函数进行调用的时候，只需要计算一次，之后的调用可以直接从缓存中返回计算结果。

使用下面的httpGetBody()作为我们需要缓存的函数样例。

func httpGetBody(url string) (interface{}, error) {
        resp, err := http.Get(url)
        if err != nil {
                return nil, err
        }
        defer resp.Body.Close()
        return ioutil.ReadAll(resp.Body) // ReadAll会返回两个结果，一个[]byte数组和一个错误
}

要求

缓存的设计要求是并发安全的，并且要尽量高效。

版本1：使用互斥量实现并发安全

版本1

// Func 是待缓存的函数（即key）
type Func func(key string) (interface{}, error)
// Result 作为缓存结果（即value）
type result struct {
        value interface{}
        err error
}
// 缓存通过调用 f 函数得到的结果
type Memo struct {
        f Func
        cache map[string]result
}

func NewMemo(f Func) *Memo {
        memo := &Memo{f, make(map[string]result)}
        return memo
}

// Get方法，线程不安全
func (memo *Memo) Get(url string) (interface{}, error) {
        res, ok := memo.cache[url]
        if !ok { // 如果缓存中不存在，通过调用memo中的f函数计算出结果，并把结果缓存起来
                res.value, res.err = memo.f(url)
                memo.cache[url] = res
        }
        return res.value, res.err
}

Memo实例会记录需要缓存的函数f(类型为Func)，以及缓存内容(里面是一个string到result映射的map)。

这是一个最简单的实现，由于没有加锁，是线程不安全的。我们先对其进行简单的测试。测试函数如下：

var urls = []string {
        "https://www.nowcoder.com/",
        "https://www.nowcoder.com/contestRoom",
        "https://www.nowcoder.com/interview/ai/index",
        "https://www.nowcoder.com/courses",
        "https://www.nowcoder.com/recommend",
        "https://www.nowcoder.com/courses",     // 重复的url，测试缓存效果
        "https://www.nowcoder.com/contestRoom", // 重复的url，测试缓存效果
}

// 单个goroutine，顺序调用
func TestMemoSingle(t *testing.T) {
        m := NewMemo(httpGetBody)
        totalTime := time.Now()
        for _, url := range urls {
                start := time.Now()
                value, err := m.Get(url)
                if err != nil {
                        log.Println(err)
                }
                fmt.Printf("%s, %s, %d bytes\n", url, time.Since(start), len(value.([]byte)))
        }
        fmt.Printf("total time used: %s\n", time.Since(totalTime))
}

// 并发调用
// 使用 sync.WaitGroup 来等待所有的请求都完成再返回
func TestMemoConcurrency(t *testing.T) {
        m := NewMemo(httpGetBody)
        var group sync.WaitGroup
        totalTime := time.Now()
        for _, url := range urls {
                group.Add(1)

                go func(url string) {
                        start := time.Now()
                        value, err := m.Get(url)
                        if err != nil {
                                log.Println(err)
                        }
                        fmt.Printf("%s, %s, %d bytes\n", url, time.Since(start), len(value.([]byte)))

                        group.Done() // equals ==> group.Add(-1)
                }(url)
        }
        group.Wait()
        fmt.Printf("total time used: %s\n", time.Since(totalTime))
}

首先测试单个goroutine顺序执行的情况，测试结果如下：

$ go test -v -run=TestMemoSingle
=== RUN   TestMemoSingle
https://www.nowcoder.com/, 289.8287ms, 95378 bytes
https://www.nowcoder.com/contestRoom, 178.8973ms, 71541 bytes
https://www.nowcoder.com/interview/ai/index, 68.9602ms, 21320 bytes
https://www.nowcoder.com/courses, 148.9146ms, 64304 bytes
https://www.nowcoder.com/recommend, 121.932ms, 90666 bytes
https://www.nowcoder.com/courses, 0s, 64304 bytes     // 可以看到，本次调用直接从缓存中获取结果，耗时为0
https://www.nowcoder.com/contestRoom, 0s, 71541 bytes // 同上
total time used: 809.5305ms
--- PASS: TestMemoSingle (0.81s)
PASS
ok      _/D_/workspace/GoRepo/gopl/ch9/memo1    1.546s

可以清楚的看到，当访问之前已经被访问过的 url 时，可以立刻从缓存中返回结果。我们再来试试看并发访问的情况。

$ go test -v -run=TestMemoConcurrency
=== RUN   TestMemoConcurrency
https://www.nowcoder.com/interview/ai/index, 252.8542ms, 21320 bytes
https://www.nowcoder.com/, 253.8524ms, 95378 bytes
https://www.nowcoder.com/recommend, 279.8401ms, 90666 bytes
https://www.nowcoder.com/courses, 280.8377ms, 64304 bytes
https://www.nowcoder.com/courses, 318.8194ms, 64304 bytes
https://www.nowcoder.com/contestRoom, 359.7913ms, 71541 bytes
https://www.nowcoder.com/contestRoom, 404.7649ms, 71541 bytes
total time used: 404.7649ms
--- PASS: TestMemoConcurrency (0.40s)
PASS
ok      _/D_/workspace/GoRepo/gopl/ch9/memo1    3.034s

并发访问时（请多测试几次），可以看到，总的用时比单个gouroutine顺序访问时少了差不多一半。但访问相同 url 时似乎没有达到缓存的效果。原因很简单嘛，我们在实现Get()方法时，没有加锁限制，因此多个goroutine可能同时访问memo实例，也就是出现了数据竞争。

在 Go 中，我们可以利用-race标签，它能帮助我们识别代码中是否出现了数据竞争。比如：

$ go test -v -race -run=TestMemoConcurrency
=== RUN   TestMemoConcurrency
...
==================
WARNING: DATA RACE
Write at 0x00c000078cc0 by goroutine 10: // 在 goroutine 10 中写入
  runtime.mapassign_faststr()
      D:/soft/Go/src/runtime/map_faststr.go:202 +0x0
  _/D_/workspace/GoRepo/gopl/ch9/memo1.(*Memo).Get()
      D:/workspace/GoRepo/gopl/ch9/memo1/memo.go:40 +0x1d5
  _/D_/workspace/GoRepo/gopl/ch9/memo1.TestMemoConcurrency.func1()
      D:/workspace/GoRepo/gopl/ch9/memo1/memo_test.go:46 +0x96

Previous write at 0x00c000078cc0 by goroutine 7: // 在 goroutine 7 中也出现写入
  runtime.mapassign_faststr()
      D:/soft/Go/src/runtime/map_faststr.go:202 +0x0
  _/D_/workspace/GoRepo/gopl/ch9/memo1.(*Memo).Get()
      D:/workspace/GoRepo/gopl/ch9/memo1/memo.go:40 +0x1d5
  _/D_/workspace/GoRepo/gopl/ch9/memo1.TestMemoConcurrency.func1()
      D:/workspace/GoRepo/gopl/ch9/memo1/memo_test.go:46 +0x96
...

FAIL
exit status 1
FAIL    _/D_/workspace/GoRepo/gopl/ch9/memo1    0.699s

可以看到，memo.go 的第40行（对应memo.cache[url] = res）出现了2次，说明有两个goroutine在没有同步干预的情况下更新了cache map。这表明Get不是并发安全的，存在数据竞争。

OK，那我们就设法对Get()方法进行加锁(mutex)，最粗暴的方式莫过于如下：

// Get is concurrency-safe.
func (memo *Memo) Get(key string) (value interface{}, err error) {
    memo.mu.Lock()
    res, ok := memo.cache[key]
    if !ok {
        res.value, res.err = memo.f(key)
        memo.cache[key] = res
    }
    memo.mu.Unlock()
    return res.value, res.err
}

这样做当然实现了所谓的“并发安全”，但是也失去了“并发性”，每次对f的调用期间都会持有锁，Get将本来可以并行运行的I/O操作串行化了。显然，这不是我们所希望的。

我们试图降低锁的粒度，查找阶段获取一次，如果查找没有返回任何内容，那么进入更新阶段会再次获取。在这两次获取锁的中间阶段，其它goroutine可以随意使用cache。

func (memo *Memo) Get(key string) (value interface{}, err error) {
    memo.mu.Lock()
    res, ok := memo.cache[key]
    memo.mu.Unlock()
    if !ok {
        res.value, res.err = memo.f(key)

        // Between the two critical sections, several goroutines
        // may race to compute f(key) and update the map.
        memo.mu.Lock()
        memo.cache[key] = res
        memo.mu.Unlock()
    }
    return res.value, res.err
}

这种实现在两个以上的goroutine同一时刻调用Get来请求同样的URL时，会导致同样的url被重复计算。多个goroutine一起查询cache，发现没有值，然后一起调用f这个慢不拉叽的函数。在得到结果后，也都会去更新map。其中一个获得的结果会覆盖掉另一个的结果。理想情况下是应该避免掉多余的工作的，这种“避免”工作一般被称为duplicate suppression(重复抑制/避免)。

版本2：使用“互斥量+channel”实现并发安全机制

该版本的Memo每一个map元素都是指向一个条目的指针。每一个条目包含对函数f的调用结果。与之前不同的是这次entry还包含了一个叫ready的channel。在条目的结果被设置之后，这个channel就会被关闭，以向其它goroutine广播——“现在去读取该条目内的结果是安全的了”。

// Func 是待缓存的函数，作为 key
type Func func(key string) (interface{}, error)

// entry 作为缓存的 value, 除了包含一个结果result，还包含一个channel
type entry struct {
        res result
        ready chan struct{}
}

type result struct {
        value interface{}
        err   error
}

// 缓存通过调用 f 函数得到的结果
type Memo struct {
        f     Func
        mu    sync.Mutex
        cache map[string]*entry
}

func NewMemo(f Func) *Memo {
        memo := &Memo{f: f, cache: make(map[string]*entry)}
        return memo
}

// 使用一个互斥量(即 unbuffered channel)来保护多个goroutine调用Get时的共享map变量
func (memo *Memo) Get(url string) (interface{}, error) {
        memo.mu.Lock()
        e := memo.cache[url]
        if e == nil {
                // 如果查询结果为空，说明这是对该url的第一次查询
                // 因此，让这个goroutine负责计算这个url对应的值
                // 当计算好后，再广播通知所有的其他goroutine，
                // 告诉它们这个url对应的缓存已经存在了，可以直接取用
                e = &entry{ready: make(chan struct{})}
                memo.cache[url] = e // 注意这里只是存入了一个“空的”条目，真正的结果还没计算出来
                memo.mu.Unlock()

                e.res.value, e.res.err = memo.f(url)

                close(e.ready) // broadcast ready condition
        } else {
                // 如果查询到结果非空，则立马先把锁给释放掉
                memo.mu.Unlock()

                // 但是要注意，这里的非空并不代表马上就可以返回结果
                // 因为有可能是其他goroutine还在计算中
                // 因此要等待ready condition
                <-e.ready
        }
        return e.res.value, e.res.err
}

获取互斥锁来保护共享变量cache map，查询map中是否存在指定条目，如果没有找到，那么分配空间插入一个新条目，释放互斥锁。如果条目存在但其值并没有写入完成时(也就是有其它的goroutine在调用 f 这个慢函数)，goroutine则必须等待ready之后才能读到条目的结果。ready condition由一个无缓存channel来实现，对无缓存channel的读取操作（即<-e.ready）在channel关闭之前一直是阻塞。

如果没有条目的话，需要向map中插入一个没有准备好的条目，当前正在调用的goroutine就需要负责调用慢函数、更新条目以及向其它所有goroutine广播条目已经ready可读的消息了。

条目中的e.res.value和e.res.err变量是在多个goroutine之间共享的。创建条目的goroutine同时也会设置条目的值，其它goroutine在收到"ready"的广播消息之后立刻会去读取条目的值。尽管会被多个goroutine同时访问，但却并不需要互斥锁。ready channel的关闭一定会发生在其它goroutine接收到广播事件之前，因此第一个goroutine对这些变量的写操作是一定发生在这些读操作之前的。不会发生数据竞争。

版本3：通过goroutine通信实现并发安全

在版本2的实现中，我们使用了一个互斥量来保护多个goroutine调用Get时的共享变量map。在Go中，还有另外一种设计方案——把共享变量map限制在一个单独的goroutine中（我们称这样的goroutine为monitor goroutine），对缓存的查询和写入均通过monitor goroutine进行。

Func、result和entry的声明和之前保持一致，这里不再重复。Memo类型的定义则做了很大的改动，只包含了一个叫做requests的channel，Get的调用者用这个channel来和monitor goroutine通信。

type request struct {
        url string
        // 负责发送响应结果, 只发送, 不接收
        response chan<- result 
}

type Memo struct {
        requests chan request
}

func NewMemo(f Func) *Memo {
        memo := &Memo{requests:make(chan request)}
        go memo.server(f)
        return memo
}

func (memo *Memo) Get(url string) (interface{}, error) {
        response := make(chan result)
        memo.requests <- request{url, response}
        res := <-response
        return res.value, res.err
}

func (memo *Memo) Close() {
        close(memo.requests)
}

上面的Get方法，会创建一个response channel，把它放进request结构中，然后发送给monitor goroutine，然后马上又会接收它。

cache变量被限制在了monitor goroutine中，即server()函数，下面会看到。monitor会在循环中一直读取请求，直到request channel被Close方法关闭。每一个请求都会去查询cache，如果没有找到条目的话，那么就会创建/插入一个新的条目。

func (memo *Memo) server(f Func) {
        cache := make(map[string]*entry)
        for req := range memo.requests {
                e := cache[req.url]
                if e == nil {
                        // this is the first request for this url
                        e = &entry{ready: make(chan struct{})}
                        cache[req.url] = e
                        go e.call(f,req.url)
                }
                go e.deliver(req.response)
        }
}

func (e *entry) call(f Func, url string) {
        // Evaluate the function.
        e.res.value, e.res.err = f(url)
        // broadcast ready condition
        close(e.ready)
}

func (e *entry) deliver(response chan<- result) {
        // wait for the ready condition
        <-e.ready
        // send the result to the client
        response <- e.res
}

和基于互斥量的版本类似，第一个对某个key的请求需要负责去调用函数f并传入这个key，将结果存在条目里，并关闭ready channel来广播条目的ready消息。使用(*entry).call来完成上述工作。

紧接着对同一个key的请求会发现map中已经有了存在的条目，然后会等待结果变为ready，并将结果从response发送给客户端的goroutien。上述工作是用(*entry).deliver来完成的。对call和deliver方法的调用必须让它们在自己的goroutine中进行以确保monitor goroutines不会因此而被阻塞住而没法处理新的请求。

总结

在Go中，我们可以通过使用互斥量（加锁），或者通信来建立并发程序。后者实现起来会难一些，初学也比较难理解。我也理解不深，暂记录于此。

本文是对《The Go Programming Language》 9.7 节的学习笔记，大家去看原文吧~

上一篇 »PHP的线程安全ZTS与非线程，NTS安全版本的区别
下一篇 »PHP----------线程安全和非线程安全的介绍

Go实现线程安全的缓存

场景

要求

版本1：使用互斥量实现并发安全

版本2：使用“互斥量+channel”实现并发安全机制

版本3：通过goroutine通信实现并发安全

总结

相关推荐

Java线程安全和非线程安全

匿名内部类创建线程对象；Thread类中的常用方法；多线程中的线程安全问题及处理方法，Java Day22

java 之前的安全的类回顾，以及以后需要线程安全时使用哪些类？

Java中常用集合效率比较

java如何解决线程安全问题？

java中volatile不能保证线程安全

GO Channel并发、死锁问题

go并发