秒杀商品,【Go语言实战】 (13) 商品秒杀的本质以及Golang实现解决方案
文章目录
写在前面
这是一个关于Go语言实现商城秒杀的解决方案。
其实商城秒杀是一个高并发问题。高并发下,我们主要解决数据竞争问题。
源代码:
当两个或多个协程同时访问同一个内存地址时发生数据竞争,并且其中至少有一个正在写入。比如线程A修改后,线程B读取线程A之前的值(初始值),所以不知道A是否被修改,所以会导致线程B把自己修改后的值放到这个内存地址,会导致这个修改没有意义。
常用的方法是加锁,当进程执行完毕后,加锁进程,防止其他进程修改数据秒杀商品,所以数据修改后释放锁。
关于锁,我们有两种锁机制,悲观锁和乐观锁。
1. 场景描述 1.1 场景描述
在这个秒杀商城中,我们对数据库中的商品数量进行操作。
秒杀产品
秒杀列表
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/不带 -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个锁(同步包中的Mutex类型互斥锁),没问题
p>
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锁(数据库乐观锁秒杀商品,【Go语言实战】 (13) 商品秒杀的本质以及Golang实现解决方案,正常)
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个使用通道限制,正常
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])
}
}
}
p>
把每个product id和user id都放进去,然后把channel当做锁,起到阻塞的作用。
3.分布式3.1环境搭建三主三从Redis集群集群模式,配置Redisson搭建ETCD集群3.2实现方式3. 2.1 基于Redisson的Redis分布式锁,正常
api/v2/with-redission?gid=1197
小心提交整个事务秒杀商品,【Go语言实战】 (13) 商品秒杀的本质以及Golang实现解决方案,并使用 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消息队列的分布式锁
改进:
评论前必须登录!
注册