互斥锁

使用一个互斥锁来在 Go 协程间安全的访问数据。

package main

import (
    "fmt"
    "math/rand"
    "runtime"
    "sync"
    "sync/atomic"
    "time"
)

func main() {

    // 在我们的例子中，`state` 是一个 map。
    var state = make(map[int]int)

    // 这里的 `mutex` 将同步对 `state` 的访问。
    var mutex = &sync.Mutex{}

    // 为了比较基于互斥锁的处理方式和我们后面将要看到的其他
    // 方式，`ops` 将记录我们对 state 的操作次数。
    var ops int64 = 0

    // 这里我们运行 100 个 Go 协程来重复读取 state。
    for r := 0; r < 100; r++ {
        go func() {
            total := 0
            for {
                // 每次循环读取，我们使用一个键来进行访问，
                // `Lock()` 这个 `mutex` 来确保对 `state` 的
                // 独占访问，读取选定的键的值，`Unlock()` 这个
                // mutex，并且 `ops` 值加 1。
                key := rand.Intn(5)
                mutex.Lock()
                total += state[key]
                mutex.Unlock()
                atomic.AddInt64(&ops, 1)
                // 为了确保这个 Go 协程不会在调度中饿死，我们
                // 在每次操作后使用 `runtime.Gosched()`
                // 进行释放。这个释放一般是自动处理的，像例如
                // 每个通道操作后或者 `time.Sleep` 的阻塞调用后
                // 相似，但是在这个例子中我们需要手动的处理。
                runtime.Gosched()
            }
        }()
    }

    // 同样的，我们运行 10 个 Go 协程来模拟写入操作，使用
    // 和读取相同的模式。
    for w := 0; w < 10; w++ {
        go func() {
            for {
                key := rand.Intn(5)
                val := rand.Intn(100)
                mutex.Lock()
                state[key] = val
                mutex.Unlock()
                atomic.AddInt64(&ops, 1)
                runtime.Gosched()
            }
        }()
    }

    // 让这 10 个 Go 协程对 `state` 和 `mutex` 的操作
    // 运行 1 s。
    time.Sleep(time.Second)

    // 获取并输出最终的操作计数。
    opsFinal := atomic.LoadInt64(&ops)
    fmt.Println("ops:", opsFinal)

    // 对 `state` 使用一个最终的锁，显示它是如何结束的。
    mutex.Lock()
    fmt.Println("state:", state)
    mutex.Unlock()
}

Go 状态协程

在前面的例子中，我们用互斥锁进行了明确的锁定来让共享的 state 跨多个 Go 协程同步访问。另一个选择是使用内置的 Go 协程和通道的的同步特性来达到同样的效果。这个基于通道的方法和 Go 通过通信以及每个 Go 协程间通过通讯来共享内存，确保每块数据有单独的 Go 协程所有的思路是一致的。

package main

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

// 在这个例子中，state 将被一个单独的 Go 协程拥有。这就
// 能够保证数据在并行读取时不会混乱。为了对 state 进行
// 读取或者写入，其他的 Go 协程将发送一条数据到拥有的 Go
// 协程中，然后接收对应的回复。结构体 `readOp` 和 `writeOp`
// 封装这些请求，并且是拥有 Go 协程响应的一个方式。
type readOp struct {
    key  int
    resp chan int
}
type writeOp struct {
    key  int
    val  int
    resp chan bool
}

func main() {

    // 和前面一样，我们将计算我们执行操作的次数。
    var ops int64

    // `reads` 和 `writes` 通道分别将被其他 Go 协程用来发
    // 布读和写请求。
    reads := make(chan *readOp)
    writes := make(chan *writeOp)

    // 这个就是拥有 `state` 的那个 Go 协程，和前面例子中的
    // map一样，不过这里是被这个状态协程私有的。这个 Go 协程
    // 反复响应到达的请求。先响应到达的请求，然后返回一个值到
    // 响应通道 `resp` 来表示操作成功（或者是 `reads` 中请求的值）
    go func() {
        var state = make(map[int]int)
        for {
            select {
            case read := <-reads:
                read.resp <- state[read.key]
            case write := <-writes:
                state[write.key] = write.val
                write.resp <- true
            }
        }
    }()

    // 启动 100 个 Go 协程通过 `reads` 通道发起对 state 所有者
    // Go 协程的读取请求。每个读取请求需要构造一个 `readOp`，
    // 发送它到 `reads` 通道中，并通过给定的 `resp` 通道接收结果。
    for r := 0; r < 100; r++ {
        go func() {
            for {
                read := &readOp{
                    key:  rand.Intn(5),
                    resp: make(chan int)}
                reads <- read
                <-read.resp
                atomic.AddInt64(&ops, 1)
            }
        }()
    }

    // 用相同的方法启动 10 个写操作。
    for w := 0; w < 10; w++ {
        go func() {
            for {
                write := &writeOp{
                    key:  rand.Intn(5),
                    val:  rand.Intn(100),
                    resp: make(chan bool)}
                writes <- write
                <-write.resp
                atomic.AddInt64(&ops, 1)
            }
        }()
    }

    // 让 Go 协程们跑 1s。
    time.Sleep(time.Second)

    // 最后，获取并报告 `ops` 值。
    opsFinal := atomic.LoadInt64(&ops)
    fmt.Println("ops:", opsFinal)
}