网站首页 返回列表像“草根”一样，紧贴着地面，低调的存在，冬去春来，枯荣无恙。

golang 高并发数据库库存同步处理

2020-06-10 03:00:10 admin 1394

## 方法一（对数据库的读写操作加锁）

### 一（DAO层加行锁:读写锁）

package main
    
    import (
        "sync"
    )
    //1、多个读之间不存在互斥关系
    //2、写操作之间都是互斥的，并且写操作与读操作之间也都是互斥的
    type idMutex map[string] *sync.RWMutex
    
    var myIdMutex idMutex
    func init() {
        myIdMutex=make(map[string] *sync.RWMutex)
    }
    //读锁定
    func (this *idMutex)RLock(id string){
        m,ok:=(*this)[id]
        if !ok {
            m=new(sync.RWMutex)
            (*this)[id]=m
        }
        m.RLock()
    }
    //读解锁
    func (this *idMutex)RUnlock(id string){
        m,ok:=(*this)[id]
        if !ok {
            return
        }
        m.RUnlock()
    }
    //写操作锁定
    func (this *idMutex)Lock(id string){
        m,ok:=(*this)[id]
        if !ok {
            m=new(sync.RWMutex)
            (*this)[id]=m
        }
        m.Lock() //写操作锁定
    }
    //写操作解锁
    func (this *idMutex)Unlock(id string) {
        m,ok:=(*this)[id]
        if !ok {
            return
        }
        m.Unlock()//写操作解锁
    }
    type Accout struct {
        Id string
    }
    //进行读的操作
    func (this *Accout) Reed() {
        myIdMutex.RLock(this.Id)
        defer myIdMutex.RUnlock(this.Id)
    }
    //进行写的操作
    func (this *Accout) Write() {
        myIdMutex.Lock(this.Id) //写操作锁定
        defer myIdMutex.Unlock(this.Id)  //写操作解锁
    }
    func main() {
        acout:=Accout{Id:"798456"}
        acout.Reed()
        acout.Write()
    }

### 一（对象加锁） 将读写的方法封装，并且添加锁

type Accout struct {
        flag sync.Mutex            //sync.Mutex类型
    }
    //进行读的操作
    func (a *Accout) Reed(n int) {  //读
        a.flag.Lock()            //锁上
        defer a.flag.Unlock()    //在方法运行完之后解开
    }
    //进行写的操作
    func (a *Accout) Write(n int) {  //读
        a.flag.Lock()            //锁上
        defer a.flag.Unlock()    //在方法运行完之后解开
    }

## 方法二（直接对数据库进行操作）

### 原理

数据库使用InnoDB

__ InnoDB与MyISAM的最大不同有两点：一是支持事务（TRANSACTION）；二是采用了行级锁。

__ InnoDB实现了以下两种类型的行锁。

* 共享锁（s）：允许一个事务去读一行，阻止其他事务获得相同数据集的排他锁。
  * 排他锁（Ｘ）：允许获取排他锁的事务更新数据，阻止其他事务取得相同的数据集共享读锁和排他写锁。  
另外，为了允许行锁和表锁共存，实现多粒度锁机制，InnoDB还有两种内部使用的意向锁（Intention Locks），这两种意向锁都是表锁。

* 意向共享锁（IS）：事务打算给数据行共享锁，事务在给一个数据行加共享锁前必须先取得该表的IS锁。
  * 意向排他锁（IX）：事务打算给数据行加排他锁，事务在给一个数据行加排他锁前必须先取得该表的IX锁。

__ 如果一个事务请求的锁模式与当前的锁兼容，InnoDB就请求的锁授予该事务；反之，如果两者两者不兼容，该事务就要等待锁释放。

__
意向锁是InnoDB自动加的，不需用户干预。对于UPDATE、DELETE和INSERT语句，InnoDB会自动给涉及及数据集加排他锁（Ｘ）（允许获取排它锁的事物进行操作，其他事物处于阻塞状态）；对于普通SELECT语句，InnoDB不会任何锁；事务可以通过以下语句显示给记录集加共享锁或排锁。

共享锁（Ｓ）：SELECT * FROM table_name WHERE ... LOCK IN SHARE MODE
    排他锁（X）：SELECT * FROM table_name WHERE ... FOR UPDATE

### 代码

package main
    import(
        "database/sql"
        _"github.com/go-sql-driver/mysql"
        "log"
        "time"
        "math/rand"
    )
    // 连接池大小
    var MAX_POOL_SIZE = 20
    var dbPoll chan *sql.DB
    
    const (
        user="root"
        pass="root"
        db="school"
    
    )
    func putDB(db *sql.DB) {
        // 基于函数和接口间互不信任原则，这里再判断一次，养成这个好习惯哦
        if dbPoll == nil {
            dbPoll = make(chan *sql.DB, MAX_POOL_SIZE)
        }
        if len(dbPoll) >= MAX_POOL_SIZE {
            db.Close()
            return
        }
        dbPoll <- db
    }
    func initDB() {
        // 缓冲机制，相当于消息队列
        if len(dbPoll) == 0 {
            // 如果长度为0，就定义一个redis.Conn类型长度为MAX_POOL_SIZE的channel
            dbPoll = make(chan *sql.DB, MAX_POOL_SIZE)
            go func() {
                for i := 0; i < MAX_POOL_SIZE/2; i++ {
                    db,err:=sql.Open("mysql",user+":"+pass+"@tcp(localhost:3306)/"+db+"?charset=utf8")
                    if err!=nil {
                        log.Println(err)
                    }
                    putDB(db)
                }
            } ()
        }
    }
    func GetDB()  *sql.DB {
        //如果为空就初始化或者长度为零
        if dbPoll == nil||len(dbPoll) == 0{
            initDB()
        }
        return <- dbPoll
    }
    func main(){
        r := rand.New(rand.NewSource(time.Now().UnixNano()))
        for i:=0;i<10 ;i++  {
            go changeCount(r.Intn(50))
            go changeCount(r.Intn(50))
            go changeCount(r.Intn(50))
            go changeCount(r.Intn(50))
        }
        time.Sleep(3*time.Second)
    }
    func changeCount(num int )  {
        db:=GetDB()
        tx, err := db.Begin()//打开事物
        defer tx.Commit()//事物提交
        //意向共享锁（IS）：事务打算给数据行共享锁，事务在给一个数据行加共享锁前必须先取得该表的IS锁。
        //意向排他锁（IX）：事务打算给数据行加排他锁，事务在给一个数据行加排他锁前必须先取得该表的IX锁。
        //意向锁是InnoDB自动加的，不需用户干预。
        res,_ := tx.Exec("UPDATE product  set count=count-? WHERE Id=1 AND count>=?  ",num,num)
        RowsAffected, err := res.RowsAffected()
        if err != nil {
            log.Println("res.RowsAffected==================Err")
        }
        if RowsAffected>0 {
            addToOrder()
            log.Println(time.Now(),"抢购成功==================",num)
        }else {
            log.Println(time.Now(),"抢购失败==================",num)
        }
    }
    //添加到订单等操作
    func addToOrder()  {
    
    }

## 方法三（中间使用redis进行缓存）

原理可以参考[奔跑的Man这篇博客](https://link.jianshu.com?t=https%3A%2F%2Fwww.cnblogs.com%2Fiforever%2Fp%2F5796902.html)  
或者我复制他的Redis多并发问题

import (
        "github.com/garyburd/redigo/redis"
        "fmt"
        "time"
        "strconv"
        "runtime"
    )
    
    var Address = "127.0.0.1:6379"
    var Network = "tcp"
    func GetRedis()  redis.Conn  {
        c, err := redis.Dial(Network, Address)
        if err != nil {
            return GetRedis()
        }
        return c
    }
    
    func main() {
        runtime.GOMAXPROCS(runtime.NumCPU())
        for i:=0;i<100;i++  {
            go do()
            go do()
            go do()
            go do()
            go do()
            go do()
            go do()
            go do()
            go do()
            go do()
            go do()
            go do()
            go do()
            go do()
            go do()
            go do()
            go do()
            go do()
            go do()
            go do()
            go do()
            go do()
            go do()
            go do()
            go do()
            go do()
            go do()
            go do()
            go do()
            go do()
        }
        time.Sleep(time.Second*60*10)
    }
    func do()  {
        cnn:=GetRedis()
        defer   cnn.Close()
        redisLock("lock.foo",cnn,20,doFunc,"致远")
    
    }
    //lockKey锁的名称
    //cnn       redis.Conn
    //deadTime      锁默认消亡时间
    //doFunc        参数名称
    //param     方法参数
    func redisLock(lockKey string,cnn redis.Conn,deadTime int,doFunc func(interface{}),param interface{})  {
        setnxTime:=time.Now().UTC().UnixNano()
        ex,err:=cnn.Do("SETNX",lockKey,setnxTime+int64(deadTime))
        if err==nil {
            if ex==int64(0) {
                //fmt.Println("存在锁:下来判断锁是否过期了")
                lock2,err:=cnn.Do("GET",lockKey)
                if lock2==nil {
                    //fmt.Println("lock2=======为空ex",ex,ex==int64(0))
                    redisLock(lockKey ,cnn ,deadTime ,doFunc ,param )
                    return
                }
                if err!=nil {
                    redisLock(lockKey ,cnn ,deadTime ,doFunc ,param )
                    return
                }
                getTime, err :=strconv.ParseInt(string(lock2.([]uint8)), 10, 64)
                if getTime>setnxTime {
                    //锁未过期
                    //fmt.Println("锁没有过期：继续等吧")
                    redisLock(lockKey ,cnn ,deadTime ,doFunc ,param )
                    return
                }else {
                    //锁已经过期
                    time.Sleep(time.Millisecond*time.Duration(deadTime))//线程休眠
                    getsettime:=time.Now().UTC().UnixNano()
                    lock3,err:=cnn.Do("GETSET",lockKey,getsettime)
                    if lock3==nil {
                        //fmt.Println("lock3=======为空")
                        redisLock(lockKey ,cnn ,deadTime ,doFunc ,param )
                        return
                    }
                    getSetTime, err :=strconv.ParseInt(string(lock3.([]uint8)), 10, 64)
                    if err!=nil {
                        //fmt.Println("出问题了：去继续等吧")
                        redisLock(lockKey ,cnn ,deadTime ,doFunc ,param )
                        return
                    }
                    if getSetTime==getTime {//如果更改前的时间和已经过期的时间相同
                        //获得锁直接操作数据
                        //fmt.Println("锁过期：处理了死锁，可以直接操作数据")
                        doFunc(param)
                        cnn.Do("DEL",lockKey)//删除锁
                        return
                    }else{//更改前的时间和已经过期的时间不同
                        //fmt.Println("判断后：没有死锁，继续等吧")
                        redisLock(lockKey ,cnn ,deadTime ,doFunc ,param )
                        return
                    }
                }
            }else{
                //fmt.Println("不存在锁：可以操作数据")
                doFunc(param)
                cnn.Do("DEL",lockKey)//删除锁
                return
            }
        }else {
            redisLock(lockKey ,cnn ,deadTime ,doFunc ,param )
            return
        }
    }
    var count=0
    func doFunc(str interface{})  {
        count+=1
        fmt.Println("操作数据中.............============================",count,str)
        return
    }

转载文章，原文链接： golang 高并发数据库库存同步处理

关键字词golang 并发

分享到：

上一篇：golang 生成RSA公钥和私钥文件

下一篇：golang继承与接口

如需留言，请登录，没有账号？请注册

0 条评论 0 人参与

网站首页 返回列表像“草根”一样，紧贴着地面，低调的存在，冬去春来，枯荣无恙。

golang 高并发数据库库存同步处理

admin 1年前

zhou 2年前

zhou 2年前

zhou 2年前

zhou 2年前

网站首页 返回列表 像“草根”一样，紧贴着地面，低调的存在，冬去春来，枯荣无恙。

golang 高并发数据库库存同步处理

admin 官方 1年前

zhou 2年前

zhou 2年前

zhou 2年前

zhou 2年前

网站首页返回列表像“草根”一样，紧贴着地面，低调的存在，冬去春来，枯荣无恙。

admin 1年前