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

秒杀商品,【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消息队列的分布式锁

改进：