Go并发编程：使用RWMutex实现高效的读写互斥

package main

import (
    "log"
    "math/rand"
    "time"
)

// ReqType 定义请求类型
type ReqType int

const (
    READ  ReqType = iota // 读请求
    WRITE                // 写请求
)

// DbRequest 数据库请求结构
type DbRequest struct {
    Type  ReqType          // 请求类型
    RespC chan *DbResponse // 响应通道
    Key   int              // 示例：请求的键
    Value string           // 示例：写请求的值
}

// DbResponse 数据库响应结构
type DbResponse struct {
    Result string // 示例：操作结果
    Found  bool   // 示例：读操作是否找到
}

// Db 模拟数据库结构
type Db struct {
    // 实际数据存储，例如 map[int]string
    data map[int]string
}

// randomWait 模拟耗时操作
func randomWait() {
    time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)
}

// readsHandler 负责处理读请求的Goroutine
func (d *Db) readsHandler(in <-chan *DbRequest) {
    for r := range in {
        // 模拟读操作
        log.Printf("Read %d starts", r.Key)
        randomWait()
        // 实际应从d.data读取
        value, ok := d.data[r.Key]
        r.RespC <- &DbResponse{Result: value, Found: ok}
        log.Printf("Read %d ends", r.Key)
    }
}

// writesHandler 负责处理写请求的函数
func (d *Db) writesHandler(r *DbRequest) *DbResponse {
    // 模拟写操作
    log.Printf("Write %d starts", r.Key)
    randomWait()
    // 实际应写入d.data
    d.data[r.Key] = r.Value
    log.Printf("Write %d ends", r.Key)
    return &DbResponse{Result: "success"}
}

// Start 启动数据库引擎
func (d *Db) Start(nReaders int) chan *DbRequest {
    in := make(chan *DbRequest, 100) // 主请求通道
    reads := make(chan *DbRequest, nReaders) // 读请求分发通道

    // 初始化数据
    d.data = make(map[int]string)
    for i := 0; i < 5; i++ {
        d.data[i] = "initial_value_" + string(rune('A'+i))
    }

    // 启动多个读Goroutine
    for k := 0; k < nReaders; k++ {
        go d.readsHandler(reads)
    }

    // 核心调度Goroutine
    go func() {
        for r := range in {
            switch r.Type {
            case READ:
                // 将读请求发送给任意一个读Goroutine
                reads <- r
            case WRITE:
                // 问题所在：如何确保在执行写操作时，所有正在进行的读操作都已完成？
                // 并且在写操作期间，不再有新的读操作开始？
                // 这是一个通过通道难以优雅解决的复杂协调问题。
                // 当前实现无法保证写操作的独占性。

                // 假设writesHandler是阻塞的，这只能防止新的读请求在当前写请求完成前被添加到`reads`通道，
                // 但不能停止或等待已经分发出去的读请求。
                r.RespC <- d.writesHandler(r)
            }
        }
    }()

    return in
}

func main() {
    rand.Seed(time.Now().UnixNano())

    blackhole := make(chan *DbResponse, 100) // 用于接收响应，不处理

    d := Db{}
    reqChannel := d.Start(4) // 启动4个读Goroutine

    go func() {
        for {
            <-blackhole // 消费响应，避免阻塞
        }
    }()

    // 模拟并发读写请求
    for i := 0; i < 20; i++ {
        key := rand.Intn(5) // 操作键0-4
        if rand.Intn(2) == 0 { // 50%概率读
            reqChannel <- &DbRequest{Type: READ, RespC: blackhole, Key: key}
        } else { // 50%概率写
            reqChannel <- &DbRequest{Type: WRITE, RespC: blackhole, Key: key, Value: "new_value_" + time.Now().Format("150405")}
        }
        time.Sleep(time.Duration(rand.Intn(50)) * time.Millisecond)
    }
    time.Sleep(2 * time.Second) // 等待一些请求完成
    log.Println("主程序退出")
}

package main

import (
    "log"
    "math/rand"
    "sync" // 引入sync包
    "time"
)

// Db 内存数据库结构，嵌入RWMutex
type Db struct {
    sync.RWMutex         // 嵌入读写锁
    data         map[int]string // 实际数据存储
}

// NewDb 创建一个新的Db实例
func NewDb() *Db {
    db := &Db{
        data: make(map[int]string),
    }
    // 初始化一些数据
    for i := 0; i < 5; i++ {
        db.data[i] = "initial_value_" + string(rune('A'+i))
    }
    return db
}

// Read 方法：获取读锁，执行读操作
func (d *Db) Read(key int) (string, bool) {
    d.RLock() // 获取读锁
    defer d.RUnlock() // 确保读锁被释放

    log.Printf("Read %d starts", key)
    time.Sleep(time.Duration(rand.Intn(50)) * time.Millisecond) // 模拟读操作耗时
    value, ok := d.data[key]
    log.Printf("Read %d ends, value: %s, found: %t", key, value, ok)
    return value, ok
}

// Write 方法：获取写锁，执行写操作
func (d *Db) Write(key int, value string) {
    d.Lock() // 获取写锁
    defer d.Unlock() // 确保写锁被释放

    log.Printf("Write %d starts", key)
    time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond) // 模拟写操作耗时
    d.data[key] = value
    log.Printf("Write %d ends, new value: %s", key, value)
}

func main() {
    rand.Seed(time.Now().UnixNano())

    db := NewDb()
    var wg sync.WaitGroup // 用于等待所有Goroutine完成

    // 启动多个Goroutine模拟并发读写请求
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func(workerID int) {
            defer wg.Done()
            for j := 0; j < 5; j++ { // 每个Goroutine执行5次操作
                key := rand.Intn(5) // 操作键0-4
                if rand.Intn(2) == 0 { // 50%概率执行读操作
                    db.Read(key)
                } else { // 50%概率执行写操作
                    db.Write(key, "updated_by_"+time.Now().Format("150405.000"))
                }
                time.Sleep(time.Duration(rand.Intn(20)) * time.Millisecond) // 模拟思考时间
            }
        }(i)
    }

    wg.Wait() // 等待所有Goroutine完成
    log.Println("所有并发操作完成。最终数据库状态：")
    // 打印最终状态，这里也需要读锁
    db.RLock()
    defer db.RUnlock()
    for k, v := range db.data {
        log.Printf("Key: %d, Value: %s", k, v)
    }
}