尝试实现(复制)数据库/缓存查询场景时出现 Golang (GO) channel 问题

Question

所以我是 channel 、 WaitGroup 、互斥锁等的新手，并尝试创建一个应用程序来查询结构的 slice 以获取数据，如果找到数据，则将其加载到 map 中。
我基本上是在尝试复制缓存/数据库场景(但为了便于理解，目前两者都在内存中)。

现在，在查询数据时，它会从数据库和缓存中查询，我为此设置了一个 RWMutex；但是在使用另一个 go 例程(通过 channel )读取存储到缓存或数据库中的数据时。它从 (db go-routine) 和 (cache go-routine) 中读取。所以我所做的是每次我从缓存 go-routine 中读取数据时，我都会耗尽一个元素的 db go-routing。

package main

import (
    "fmt"
    "math/rand"
    "strconv"
    "sync"
    "time"
)

type Book struct {
    id   int
    name string
}

var cache = map[int]Book{}
var books []Book
var rnd = rand.New(rand.NewSource(time.Now().UnixNano()))

func main() {
    cacheCh := make(chan Book)
    dbCh := make(chan Book)
    wg := &sync.WaitGroup{}
    m := &sync.RWMutex{}
    loadDb()
    for i := 0; i < 10; i++ {
        id := rnd.Intn(10)
        wg.Add(1)
        go func(id int, wg *sync.WaitGroup, m *sync.RWMutex, ch chan<- Book) {
            if find, book := queryCache(id, m); find {
                fmt.Println("Found Book In Cache: ", book)
                ch <- book
            }
            wg.Done()
        }(id, wg, m, cacheCh)
        wg.Add(1)
        go func(id int, wg *sync.WaitGroup, m *sync.RWMutex, ch chan<- Book) {
            if find, book := queryDb(id, m); find {
                ch <- book
            }
            wg.Done()
        }(id, wg, m, dbCh)
        go func(dbCh, cacheCh <-chan Book) {
            var book Book
            select {
            case book = <-cacheCh:
                msg := <-dbCh
                fmt.Println("Drain DbCh From: ", msg, "\nBook From Cache: ", book.name)
            case book = <-dbCh:
                fmt.Println("Book From Database: ", book.name)
            }
        }(dbCh, cacheCh)
    }

    wg.Wait()
}

func queryCache(id int, m *sync.RWMutex) (bool, Book) {
    m.RLock()
    b, ok := cache[id]
    m.RUnlock()
    return ok, b
}
func queryDb(id int, m *sync.RWMutex) (bool, Book) {
    for _, val := range books {
        if val.id == id {
            m.Lock()
            cache[id] = val
            m.Unlock()
            return true, val
        }
    }
    var bnf Book
    return false, bnf

}

func loadDb() {
    var book Book
    for i := 0; i < 10; i++ {
        book.id = i
        book.name = "a" + strconv.Itoa(i)
        books = append(books, book)
    }

}

另外，我知道在这段代码中它总是会查询数据库，即使它在缓存中发现了一个不理想的命中。但是，这只是一个测试场景，我优先考虑的是用户以可能的最快模式接收详细信息(即，如果它不存在于缓存中，它不应该等待来自缓存的响应在查询数据库之前)。

如果可能，请提供帮助，我对此很陌生，所以这可能是微不足道的。

对不起，谢谢。

Answer 1

我能看到的问题主要与关于事物执行顺序的假设有关:

您假设 db/cache 检索 go 例程将按顺序返回；不能保证这一点(理论上，请求的最后一本书可以首先添加到 channel 中，并且缓存 channel 上的条目的排序可能与 db channel 不同)。

以上观点否定了你额外的msg := <-dbCh (因为数据库可能会在缓存之前返回结果，在这种情况下， channel 不会被耗尽)。这会导致死锁。

当 WaitGroup 完成时，您的代码会退出，这可能并且可能会在检索结果的 go 例程完成之前发生(带有 select 的 go 例程不使用 WaitGroup ) - 这实际上不是问题，因为您的代码在此之前遇到死锁。

下面的代码( playground)是我尝试解决这些问题，同时让您的代码大部分保持原样(我确实创建了一个 getBook 函数，因为我认为这更容易理解)。我不会保证我的代码没有错误，但希望它会有所帮助。

package main

import (
    "fmt"
    "math/rand"
    "strconv"
    "sync"
    "time"
)

type Book struct {
    id   int
    name string
}

var cache = map[int]Book{}
var books []Book
var rnd = rand.New(rand.NewSource(time.Now().UnixNano())) // Note: Not random on playground as time is simulated

func main() {
    wg := &sync.WaitGroup{}
    m := &sync.RWMutex{}
    loadDb()
    for i := 0; i < 10; i++ {
        id := rnd.Intn(10)
        wg.Add(1) // Could also add 10 before entering the loop
        go func(wg *sync.WaitGroup, m *sync.RWMutex, id int) {
            getBook(m, id)
            wg.Done() // The book has been retrieved; loop can exit when all books retrieved
        }(wg, m, id)
    }
    wg.Wait() // Wait until all books have been retrieved
}

// getBook will retrieve a book (from the cache or the db)
func getBook(m *sync.RWMutex, id int) {
    fmt.Printf("Requesting book: %d\n", id)
    cacheCh := make(chan Book)
    dbCh := make(chan Book)
    go func(id int, m *sync.RWMutex, ch chan<- Book) {
        if find, book := queryCache(id, m); find {
            //fmt.Println("Found Book In Cache: ", book)
            ch <- book
        }
        close(ch)
    }(id, m, cacheCh)
    go func(id int, m *sync.RWMutex, ch chan<- Book) {
        if find, book := queryDb(id, m); find {
            ch <- book
        }
        close(ch)
    }(id, m, dbCh)

    // Wait for a result from one of the above - Note that we make no assumptions
    // about the order of the results but do assume that dbCh will ALWAYS return a book
    // We want to return from this function as soon as a result is received (because in reality
    // we would return the result)
    for {
        select {
        case book, ok := <-cacheCh:
            if !ok { // Book is not in the cache so we need to wait for the database query
                cacheCh = nil // Prevents select from considering this
                continue
            }
            fmt.Println("Book From Cache: ", book.name)
            drainBookChan(dbCh) // The database will always return something (in reality you would cancel a context here to stop the database query)
            return
        case book := <-dbCh:
            dbCh = nil // Nothing else will come through this channel
            fmt.Println("Book From Database: ", book.name)
            if cacheCh != nil {
                drainBookChan(cacheCh)
            }
            return
        }
    }
}

// drainBookChan starts a go routine to drain the channel passed in
func drainBookChan(bChan chan Book) {
    go func() {
        for _ = range bChan {
        }
    }()
}

func queryCache(id int, m *sync.RWMutex) (bool, Book) {
    time.Sleep(time.Duration(rnd.Intn(100)) * time.Millisecond) // Introduce some randomness (otherwise everything comes from DB)
    m.RLock()
    b, ok := cache[id]
    m.RUnlock()
    return ok, b
}
func queryDb(id int, m *sync.RWMutex) (bool, Book) {
    time.Sleep(time.Duration(rnd.Intn(100)) * time.Millisecond) // Introduce some randomness (otherwise everything comes from DB)
    for _, val := range books {
        if val.id == id {
            m.Lock()
            cache[id] = val
            m.Unlock()
            return true, val
        }
    }
    var bnf Book
    return false, bnf

}

func loadDb() {
    var book Book
    for i := 0; i < 10; i++ {
        book.id = i
        book.name = "a" + strconv.Itoa(i)
        books = append(books, book)
    }

}