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

大家好，今天本人给大家带来文章《尝试实现（复制）数据库/缓存查询场景时出现 Golang (GO) 通道问题》，文中内容主要涉及到，如果你对Golang方面的知识点感兴趣，那就请各位朋友继续看下去吧~希望能真正帮到你们，谢谢！

所以我对通道、等待组、互斥体等很陌生，并尝试创建一个应用程序来查询结构体的切片以获取数据，如果找到数据，则将其加载到映射中。 我基本上是在尝试复制缓存/数据库场景（但目前将两者都放在内存中以便于理解）。

现在，在查询数据时，它会从数据库和缓存中查询，我为此设置了一个 rwmutex；但是在使用另一个 go 例程（通过通道）读取存储到缓存或数据库中的数据时。它从 (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)
    }

}

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

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

抱歉，谢谢。

解决方案

我看到的问题主要与对事物执行顺序的假设有关：

• 您假设数据库/缓存检索 go 例程将按顺序返回；对此无法保证（理论上，最后请求的一本书可以首先添加到通道中，并且缓存通道上的条目的顺序可能与数据库通道不同）。
• 上述一点否定了额外的 msg := <-dbch （因为数据库可能会在缓存之前返回结果，在这种情况下，通道不会被耗尽）。这会导致僵局。
• 当等待组完成时，您的代码将退出，并且这可能并且可能会在检索结果的 go 例程完成之前发生（带有 select 的 go 例程不使用等待组） - 这实际上不是问题，因为您的代码在此之前陷入僵局。

下面的代码（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)
    }

}

本篇关于《尝试实现（复制）数据库/缓存查询场景时出现 Golang (GO) 通道问题》的介绍就到此结束啦，但是学无止境，想要了解学习更多关于Golang的相关知识，请关注golang学习网公众号！