【Go语言实战】 (13) 商品秒杀的本质以及Golang实现解决方案

文章目录

写在前面

这是一篇关于Go语言实现商城秒杀的解决方案。
其实商城的秒杀就是高并发问题，那高并发下我们主要解决的就是数据竞争问题

源码：https://github.com/CocaineCong/Go-SecKill

当两个或多个协程同时访问同一个内存地址，并且至少有一个是在写时，就会发生数据竞争。比如A线程修改完之后，线程B读的是线程A之前的值(初始值)，所以不知道A是否修改，所以会导致线程B也把自己修改的值放到这个内存地址中，就会导致本次修改无意义。

【Go语言实战】 (13) 商品秒杀的本质以及Golang实现解决方案_加锁
常用的方法就是加锁了，当这个进程已经执行了，就为该进程进行加锁，防止其他进程对这个数据进行修改，所以这个数据进行修改之后，再释放这个锁。

悲观锁乐观锁

保持一个数据的多版本

1. 场景说明

1.1 场景描述

本次秒杀商城，我们对数据库商品数量进行操作。

秒杀的商品
【Go语言实战】 (13) 商品秒杀的本质以及Golang实现解决方案_加锁_02
秒杀成功的名单

1.2 事务编写

初始化本次秒杀的商品

func InitializerSecKill(gid int) {
    tx := model.DB.Begin()            // 开启事务
    err := model.DeleteByGoodsId(gid) 
    // 删除前一次秒杀的所有用户，既删除表 success_killed
    if err != nil {         // 发生错误的话就进行回滚
        tx.Rollback()
    }
    err = model.UpdateCountByGoodsId(gid) 
    // 更新商品的信息表 promotion_sec_kill
    if err != nil {
        tx.Rollback()
    }
    tx.Commit()
}

开启50个线程并发进行秒杀

func WithoutLockSecKill(gid int) serializer.Response {
    code := e.SUCCESS
    seckillNum := 50
    wg.Add(seckillNum)
    InitializerSecKill(gid)
    for i := 0; i < seckillNum; i++ {
        userID := i
        go func() {
            err := WithoutLockSecKillGoods(gid, userID)
            if err != nil {
                fmt.Println("Error",err)
            } else {
                fmt.Printf("User: %d seckill successfully.\n", userID)
            }
            wg.Done()
        }()
    }
    wg.Wait()
    killedCount, err := GetKilledCount(gid)
    if err != nil {
        code = e.ERROR
        logging.Error("Seckill System Error")
        return serializer.Response{
            Status: code,
            Msg:    e.GetMsg(code),
            Error:  err.Error(),
        }
    }
    fmt.Println(killedCount)
    logging.Infof("kill %v product", killedCount)
    return serializer.Response{
        Status: code,
        Msg:    e.GetMsg(code),
    }
}

2. 单机模式

2.1 不加锁出现超卖情况

api/v1/without-lock?gid=1197

func WithoutLockSecKillGoods(gid, userID int) error {
    tx := model.DB.Begin()
    // 检查库存
    count, err := model.SelectCountByGoodsId(gid)
    if err != nil {
        return err
    }
    if count > 0 {
        // 1. 扣库存
        err = model.ReduceStockByGoodsId(gid, int(count-1))
        if err != nil {
            tx.Rollback()
            return err
        }
        // 2. 创建订单
        kill := model.SuccessKilled{
            GoodsId:    int64(gid),
            UserId:     int64(userID),
            State:      0,
            CreateTime: time.Now(),
        }
        err = model.CreateOrder(kill)
        if err != nil {
            tx.Rollback()
            return err
        }
    }
    tx.Commit()
    return nil
}

2.2 加锁(sync包中的Mutex类型的互斥锁),没有问题

api/v1/with-lock?gid=1197

func WithLockSecKillGoods(gid,userID int) error {
    lock.Lock()
    err := WithoutLockSecKillGoods(gid, userID)
    lock.Unlock()
    return err
}

2.3 加锁(数据库悲观锁，读限定), 出现超卖

api/v1/with-pcc-read?gid=1197

func SelectCountByGoodsIdPcc(gid int) (int64, error) {
    skGood:=PromotionSecKill{}
    err := DB.Model(PromotionSecKill{}).Set("gorm:query_option", "FOR UPDATE").
        Where("goods_id=?",gid).First(&skGood).Error
    return skGood.PsCount, err
}

FOR UPDATE

2.4 加锁(数据库悲观锁，更新限定), 正常

api/v1/with-pcc-update?gid=1197

func ReduceByGoodsId(gid int) (int64, error) {
    var count int64
    sqlStr := `UPDATE promotion_sec_kill SET ps_count = ps_count-1 WHERE ps_count>0 AND goods_id = ?`
    res := DB.Exec(sqlStr, gid)
    if err := res.Error; err != nil {
        return count, err
    }
    count = res.RowsAffected
    return count, nil
}

ps_count>0 进行限定作用。

2.5 加锁(数据库乐观锁，正常)

api/v1/with-occ?gid=1197

func ReduceStockByOcc(gid int, num int, version int) (int64, error) {
    var count int64
    sqlStr := "UPDATE promotion_sec_kill SET ps_count = ps_count-?, version = version+1 " +
        "WHERE version = ? AND goods_id = ?"
    res := DB.Exec(sqlStr, num, version, gid)
    if err := res.Error; err != nil {
        return count, err
    }
    count = res.RowsAffected
    return count, nil
}

使用version进行版本的控制，从而实现乐观锁。

2.6 使用 channel 限制，正常

api/v1/with-channel?gid=1197

func ChannelConsumer() {
    for {
        kill, ok := <-(*GetInstance())
        if !ok {
            continue
        }
        err := WithoutLockSecKillGoods(kill[0], kill[1])
        if err != nil {
            logging.Error("Error")
        } else {
            logging.Infof("User:%v SecKill Successfully", kill[1])
        }
    }
}

将每个商品id和用户id放入其中，然后可以把channel作为一把锁，起到了阻塞作用。

3. 分布式

3.1 环境搭建

搭建三主三从Cluster模式的Redis集群,配置Redisson
搭建ETCD集群

3.2 实现方法

3.2.1 基于Redisson的Redis分布式锁，正常

api/v2/with-redission?gid=1197

注意要用Redis Lock把整个事务提交都包住。这里仅仅使用了Redis分布式提供的锁功能，秒杀数据处理还是直接访问数据库来完成

func WithRedssionSecKillGoods(gid , userID int) error {
    g := strconv.Itoa(gid)
    uuid := getUuid(g)
    lockSuccess, err := cache.RedisClient.SetNX(g, uuid, time.Second*3).Result()
    if err != nil || !lockSuccess {
        fmt.Println("get lock fail", err)
        return errors.New("get lock fail")
    } else {
        fmt.Println("get lock success")
    }
    err = WithoutLockSecKillGoods(gid, userID)
    if err != nil {
        return err
    }
    value, _ := cache.RedisClient.Get(g).Result()
    if value == uuid { //compare value,if equal then del
        _, err := cache.RedisClient.Del(g).Result()
        if err != nil {
            fmt.Println("unlock fail")
            return nil
        } else {
            fmt.Println("unlock success")
        }
    }
    return nil
}

3.2.2 基于缓存的ETCD分布式锁，正常

api/v2/with-etcd?gid=1197

类似于之前使用BlockingQueue时编写了一个单例模式的工具类来全局使用的形式相同，注意这里也要用ETCD分布式锁把整个事务提交都包住。这里只用了ETCD的分布式锁功能，秒杀数据处理也是直接访问数据库来完成

func WithETCDSecKillGoods(gid, userID int) error {
    var conf = clientv3.Config{
        Endpoints:   []string{"127.0.0.1:2379"},
        DialTimeout: 5 * time.Second,
    }
    eMutex1 := &EtcdMutex{
        Conf: conf,
        Ttl:  10,
        Key:  "lock",
    }
    err := eMutex1.Lock()
    if err != nil {
        return err
    }
    err = WithoutLockSecKillGoods(gid, userID)
    eMutex1.UnLock()
    return err
}

3.2.3 Redis的List队列，正常

api/v2/with-redis-list?gid=1197

这里利用Redis分布式队列的方式是，在秒杀活动初始化阶段时有多少库存就在Redis的List中初始化多少个商品元素。
然后每有一个用户进行秒杀，就从List队列中取出一个商品元素分配给该用户。
同时将该用户信息存入到Redis的Set类型中，防止用户多次秒杀的情况。
在秒杀结束之后，在Redis中数据写入到数据库中进行保存。可参考下图：

func WithRedisListSecKillGoods(gid, userID int) error {
    g := strconv.Itoa(gid)
    u := strconv.Itoa(userID)
    if cache.RedisClient.Get(u + g).Val() == "" { // 这用户没有秒杀过
        cache.RedisClient.RPop(g)
        cache.RedisClient.Set(u+g, g, 3*time.Minute)
        cache.RedisClient.ZAdd(g, redis.Z{float64(time.Now().Unix()), userID})
    } else { // 这用户已经有记录了
        return errors.New("该用户已经抢过了")
    }
    return nil
}

3.2.4 Redis原子递减,正常

这里先将秒杀商品的库存数量，写入到redis中，利用redis的incr来实现原子递减。
假如有100件商品，这里相当于准备好了100个钥匙，有人没有抢到钥匙，就返回库存不够，有人抢到了钥匙，就进行下一步处理，先将秒杀订单的信息写入到redis中，等空闲下来后在写入到数据库中。这里其实与3.2.3差不多

其他

基于Redis的任务队列，订阅监听
(是将在前端进行秒杀的用户的信息传入到通道中，等待被消费。后端订阅监听这个通道，有秒杀用户信息传过来就进行消费处理，再将处理数据写入到数据库。)
基于MQ消息队列的分布式锁

改进：

索引与SQL语句检查
尽可能利用缓存
利用MQ进行流量削峰
Nginx负载均衡
读写分离与分表分库
CDN内容分发网络
流量防刷和反爬虫