Go并发模式：读写互斥中的通道与RWMutex实践

package main

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

var source *rand.Rand

type ReqType int

const (
    READ = iota
    WRITE
)

type DbRequest struct {
    Type  int              // 请求类型
    RespC chan *DbResponse // 响应通道
    // 请求内容
}

type DbResponse struct {
    // 响应内容
}

type Db struct {
    // 数据库结构体
}

func randomWait() {
    time.Sleep(time.Duration(source.Intn(1000)) * time.Millisecond)
}

func (d *Db) readsHandler(in <-chan *DbRequest) {
    for r := range in {
        id := source.Intn(4000000)
        log.Println("read ", id, " starts")
        randomWait()
        log.Println("read ", id, " ends")
        r.RespC <- &DbResponse{}
    }
}

func (d *Db) writesHandler(r *DbRequest) *DbResponse {
    id := source.Intn(4000000)
    log.Println("write ", id, " starts")
    randomWait()
    log.Println("write ", id, " ends")
    return &DbResponse{}
}

func (d *Db) Start(nReaders int) chan *DbRequest {
    in := make(chan *DbRequest, 100)
    reads := make(chan *DbRequest, nReaders) // 读者请求通道

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

    go func() {
        for r := range in {
            switch r.Type {
            case READ:
                reads <- r // 将读请求发送给读者Goroutine
            case WRITE:
                // 在这里，我们需要等待所有当前正在进行的读操作完成
                // 并且在写操作期间阻止新的读操作进入。
                // 纯粹使用通道实现这种协调逻辑会非常复杂。
                // 示例中直接执行写操作，这会导致读写冲突。
                r.RespC <- d.writesHandler(r)
            }
        }
    }()

    return in
}

func main() {
    seed := time.Now().Unix()
    source = rand.New(rand.NewSource(seed))

    blackhole := make(chan *DbResponse, 100)

    d := Db{}
    rc := d.Start(4) // 启动数据库引擎
    wc := time.After(3 * time.Second)

    go func() {
        for {
            <-blackhole // 消费响应
        }
    }()

    for {
        select {
        case <-wc:
            return // 3秒后退出
        default:
            if source.Intn(2) == 0 {
                rc <- &DbRequest{READ, blackhole} // 发送读请求
            } else {
                rc <- &DbRequest{WRITE, blackhole} // 发送写请求
            }
        }
    }
}

import "sync" // 导入sync包

type Db struct {
    sync.RWMutex // 嵌入读写互斥锁
    // 数据库数据结构，例如 map[string]interface{}
    data map[string]interface{}
}

// 示例：数据库初始化
func NewDb() *Db {
    return &Db{
        data: make(map[string]interface{}),
    }
}

// 示例：读取操作
func (d *Db) Get(key string) (interface{}, bool) {
    d.RLock() // 获取读锁
    defer d.RUnlock() // 确保读锁最终被释放
    log.Printf("Reading key: %s", key)
    time.Sleep(time.Millisecond * 50) // 模拟读取耗时
    val, ok := d.data[key]
    return val, ok
}

// 示例：写入操作
func (d *Db) Set(key string, value interface{}) {
    d.Lock() // 获取写锁
    defer d.Unlock() // 确保写锁最终被释放
    log.Printf("Writing key: %s, value: %v", key, value)
    time.Sleep(time.Millisecond * 100) // 模拟写入耗时
    d.data[key] = value
}

// 结合RWMutex的Start方法示例 (简化版，不再使用多通道协调)
func (d *Db) StartEngine() chan *DbRequest {
    in := make(chan *DbRequest, 100)

    go func() {
        for r := range in {
            switch r.Type {
            case READ:
                // 在这里调用Db的Get方法，它内部会处理RLock/RUnlock
                _, _ = d.Get("exampleKey") // 假设操作一个固定的键
                r.RespC <- &DbResponse{}
            case WRITE:
                // 在这里调用Db的Set方法，它内部会处理Lock/Unlock
                d.Set("exampleKey", rand.Intn(100)) // 假设写入一个随机值
                r.RespC <- &DbResponse{}
            }
        }
    }()
    return in
}

func main() {
    seed := time.Now().Unix()
    source = rand.New(rand.NewSource(seed))

    blackhole := make(chan *DbResponse, 100)

    db := NewDb() // 初始化数据库
    requestChannel := db.StartEngine() // 启动数据库引擎

    done := time.After(3 * time.Second)

    go func() {
        for {
            <-blackhole // 消费响应
        }
    }()

    for {
        select {
        case <-done:
            return // 3秒后退出
        default:
            if source.Intn(2) == 0 {
                requestChannel <- &DbRequest{READ, blackhole} // 发送读请求
            } else {
                requestChannel <- &DbRequest{WRITE, blackhole} // 发送写请求
            }
            time.Sleep(time.Millisecond * 10) // 控制请求发送速率
        }
    }
}

import "log"

// 示例：配置log包
func init() {
    // log.SetOutput(os.Stdout) // 如果需要输出到标准输出
    log.SetFlags(log.Ldate | log.Ltime | log.Lshortfile) // 设置日志格式，包含日期时间文件名
    // log.SetPrefix("[DB_ENGINE] ") // 设置日志前缀
}

// 之后在代码中直接使用 log.Println, log.Printf 等
// log.Println("This is a thread-safe log message.")