Go实战：抽奖系统核心技术点实现

Go 实战 &＃xff1a;抽奖系统核心技术点实现

考验系统并发能力最常见的无非是秒杀、抽奖等业务场景。我们不妨以一个支持高并发的抽奖系统的设计和实现为例来进行 Go 语言的实战学习。

要实现一个抽奖系统&＃xff0c;我们首先需要一个比较合理的抽奖算法&＃xff0c;保证每个人抽到奖品的概率是一致的&＃xff0c;同时又需要避免奖品被很快抽完的情况。所以本文的第一部分会用三种思路来讲解如何去实现一个合理的抽奖算法。

要具备高并发的能力&＃xff0c;Go 的 http 组件是支持高并发能力的(底层由 Go 的协程支持)&＃xff0c;但是数据存储层也需要支持高并发、高吞吐量的存储系统&＃xff0c;mysql 肯定是支持不了的&＃xff0c;所以这里采用 Redis 实现。所以本文第二部分会介绍如何使用 Go 去操作 Redis。

抽奖系统中&＃xff0c;高并发请求下&＃xff0c;很容易出现两个线程或者协程同时对共享内存中的数据进行更改的问题。所以为了保证数据的原子性&＃xff0c;我们对数据的操作都是以 CAS 的方式。Redis 中可以实现对数据的原子增减&＃xff0c;但是无法实现数据的翻倍或者其他改变。为了支持所有场景&＃xff0c;同时为了学习知识&＃xff0c;这里我们借用 Redis 的事务来实现 Redis 的 CAS。所以第三部分主要介绍 Go 操作 Redis 来实现 cas

整个抽奖系统概览如下&＃xff0c;本章 chat 主要阐述其中的部分实现核心点&＃xff0c;包括&＃xff1a;使用 Go 实现抽奖算法&＃xff0c;Go 操作 Redis &＃xff0c;以及通过 Redis 实现 cas

Go 实现抽奖算法

抽奖算法是决定一个抽奖系统是否成功的关键要素之一

我们先来假设一下场景&＃xff1a; 假如目前公司需要举行一个活动&＃xff0c;一共有A、B、C 三种奖品。奖品数量分别为 200、2000、5000、需要在 6.1 ~ 6.3 这三种奖品发完。

最容易想到的一种做法是对 奖品 A、B、C 分别设置一个概率&＃xff0c;每个用户根据设置好的概率进行派奖。但仔细想想这样会存在一个问题&＃xff1a;每个活动我们对于抽奖的人数预测都是未知的。当抽奖人数多或者概率很高时&＃xff0c;奖品很容易在很短的时间内被发完。当抽奖人数少或者概率很低时&＃xff0c;奖品很有可能到最后还有很多都没发完… 所以完全凭借设置抽奖概率来抽奖是不太靠谱的。

为了避免上述情况&＃xff0c;于是我们开始思考&＃xff0c;既然抽奖的人数是未知的&＃xff0c;我们无法控制抽奖人数&＃xff0c;那能不能换个思路&＃xff0c;我们来控制奖品的发放时间呢&＃xff1f;在上面假设的场景中&＃xff0c;我们分别有 A、B、C 三种奖品 200、2000、5000 个。需要在 6.1 日 0:00 ~ 6.3 日 24:00 点这段期间进行抽奖。也就是说一共有 7200 奖品&＃xff0c;在 72 个小时内抽完。假如我们把这 72 个小时按照时间片进行分割&＃xff0c;是不是就能防止奖品被抽完或者过剩太多的情况了&＃xff1f;

由公式

ΔΔ t &＃61; endTime−startTimeawardAmountendTime−startTimeawardAmount 计算可以得出时间间隔为 1/100 小时&＃xff0c;也就是 36s 随机发一个奖品

此时&＃xff0c;在 ΔΔ t 时间内 A 被抽中的概率是 200 / (200 &＃43; 2000 &＃43; 5000) &＃61; 1/36 , B 被抽中的概率是 2000 / (200 &＃43; 2000 &＃43; 5000) &＃61; 10/36 &＃xff0c;C 被抽中的概率是 5000 / (200 &＃43; 2000 &＃43; 5000) &＃61; 25 /36

算法的基本思路确定后&＃xff0c;怎么去实现呢&＃xff1f;

我们将奖品分成 7200 份之后&＃xff0c;每一个奖品都会有一个抽奖时间段&＃xff0c;在这个时间段内奖品会在某一个时间点进行发放。假如用户抽奖的时间点在这个时间点之前&＃xff0c;因为奖品没有发放&＃xff0c;所以无法中奖。在这个时间点之后&＃xff0c;则有概率获得奖品&＃xff0c;这个概率可以由用户进行配置。假如有一个用户抽中了奖品&＃xff0c;则其他用户在这个时间段内再也无法抽中奖品了。

思路一

第一种思路就是通过结构化数据库去实现&＃xff0c;例如 mysql

这里需要建立三张表

• 奖品信息表 : award_info

• • 奖品记录表 : award_record

  活动开始前&＃xff0c;需要根据奖品信息表 award_info 里面的数据初始化奖池表 award_pool。用户抽奖时通过以下 sql 进行抽奖

select id from award_pool where award_release_time <&＃61; now() and award_release_num > 0 limit 1

如果能够查询到数据&＃xff0c;则执行更新 update 语句&＃xff0c;否则抽奖失败

update award_pool set award_release_num &＃61; award_release_num - 1 where id &＃61; #{id} and award_release_num > 0

如果 update 执行成功&＃xff0c;用户中奖&＃xff0c;写中奖纪录到 award_record 表&＃xff0c;否则抽奖失败

思路二

在思路一中&＃xff0c;在抽奖之前&＃xff0c;我们对需要初始化 awardpool 这张表来进行发奖。抽奖时&＃xff0c;需要执行一条 select 语句 和 一条 update 语句。这种做法&＃xff0c;在奖品量非常多时&＃xff0c;初始化 awardpool 这张表会浪费很多时间&＃xff0c;在用户并发请求量非常大&＃xff0c;select 语句也会消耗很多性能。于是有了第二种思路&＃xff0c;不进行奖品初始化操作&＃xff0c;而是通过实时计算的方式进行抽奖。

计算奖品释放时间点 &＃xff1a;

具体实现&＃xff1a;

func GetAward(awardBatches []AwardBatch) (AwardBatch, error) {startTime , _ :&＃61; ParseStringToTime(conf.Award.StartTime)endTime , _ :&＃61; ParseStringToTime(conf.Award.EndTime)fmt.Println("startTime : ", startTime)fmt.Println("endTime : ", endTime)award , err :&＃61; RandomGetAwardBatch(awardBatches)if err !&＃61; nil {return AwardBatch{}, err}totalAmount :&＃61; award.GetTotalAmount()totalBalance :&＃61; award.GetTotalBalance()updateTime :&＃61; award.GetUpdateTime()detaTime :&＃61; (endTime - startTime) / totalAmountcurrentTime :&＃61; time.Now().Unix()random :&＃61; rand.New(rand.NewSource(updateTime))// 计算下一个奖品的释放时间releaseTime :&＃61; startTime &＃43; (totalAmount - totalBalance) * detaTime &＃43; int64(random.Int()) % detaTimefmt.Println("releaseTime : " &＃43; fmt.Sprintf("%d", releaseTime) &＃43; " currentTime : " &＃43; fmt.Sprintf("%d",currentTime))if (currentTime }func RandomGetAwardBatch(awardBatches []AwardBatch) ( AwardBatch , error ) {if len(awardBatches) &＃61;&＃61; 0 {return AwardBatch{} , errors.New("empty param awardBatches")}weight :&＃61; int64(0)for _, awardBatch :&＃61; range awardBatches {weight &＃43;&＃61; awardBatch.GetTotalBalance()}if weight &＃61;&＃61; 0 {return AwardBatch{}, errors.New("weight is 0")}random :&＃61; rand.New(rand.NewSource(weight))num :&＃61; random.Int63n(weight)for _, awardBatch :&＃61; range awardBatches {num -&＃61; awardBatch.GetTotalBalance()if num <0 {return awardBatch , nil}}return AwardBatch{}, errors.New("randomGetAwardBatch should shoot at least one batch")}


思路三

思路二相对于思路一来说&＃xff0c;节省了初始化 award_pool 奖池 和 select 操作的时间。但还是会存在一个问题&＃xff0c;由于采用的是关系型数据库 mysql 存储&＃xff0c;所以能够支持并发数是有限的。当抽奖用户量很大时&＃xff0c;数据库肯定是无法支撑的。所以为了能够支持高并发&＃xff0c;我们需要摈弃 mysql &＃xff0c;采取高性能、支持高并发量的 k/v 式数据库。这里我们使用 Redis

为了达到思路二类似的效果&＃xff0c;我们设计两种数据结构 &＃xff1a;

奖品种类、奖品总数和中奖时间采用 zset 进行存储&＃xff0c;zset 中 value 为奖品名称&＃xff0c;value 的分值 score 为奖品的中奖时间。通过 ZRANGE award_info 0 -1 WITHSCORES 即可得到所有奖品以及每种奖品的更新抽奖时间

奖品剩余数量采用哈希表进行存储&＃xff0c;这样通过 HGETALL 即可查询到所有奖品以及奖品剩余数量

基于这种思路&＃xff0c;我们就有了第三种代码实现如下&＃xff08;这里只贴部分核心代码&＃xff0c;完整代码请参考 GitHub&＃xff09;

// 随机抽取一个奖品func RandomGetAwardBatch() \*AwardBatch {conn :&＃61; RedisWrapper.GetConn()if conn &＃61;&＃61; nil {log.Println("conn is nil")return nil}defer conn.Close()retMap, err :&＃61; Redis.Int64Map(conn.Do("HGETALL", getAwardBalanceKey()))if err !&＃61; nil || retMap &＃61;&＃61; nil {log.Println("Redis HGETALL award error", err)return nil}totalBalance :&＃61; int64(0)for _, value :&＃61; range retMap {totalBalance &＃43;&＃61; value}fmt.Println("retMap : ", retMap)if totalBalance &＃61;&＃61; 0 {log.Println("total balance is 0")return nil}log.Println("totalBalance :", totalBalance)awardBatches :&＃61; GetAllAwardBatch()for index , awardBatch :&＃61; range awardBatches {awardBatches[index].totalBalance &＃61; retMap[awardBatch.GetName()]}log.Println("awardBatches :", awardBatches)random :&＃61; rand.New(rand.NewSource(totalBalance))num :&＃61; random.Int63n(totalBalance)for _ , awardBatch :&＃61; range awardBatches {// 奖品已经抽完if awardBatch.GetTotalBalance() <&＃61; 0 {log.Println("奖品已经抽完")continue}num &＃61; num - awardBatch.GetTotalBalance()if num <0 {return &awardBatch}}return nil}func GetAwardBatch() *AwardBatch {awardBatch :&＃61; RandomGetAwardBatch()if awardBatch &＃61;&＃61; nil {log.Println("sorry, you didn&＃39;t win the prize.")return nil}// 判断是否到达奖品释放时间点startTime , _ :&＃61; ParseStringToTime(Conf.Award.StartTime)endTime , _ :&＃61; ParseStringToTime(Conf.Award.EndTime)totalAmount :&＃61; awardBatch.totalAmounttotalBalance :&＃61; awardBatch.totalBalancelastUpdateTime :&＃61; awardBatch.GetUpdateTime()random :&＃61; rand.New(rand.NewSource(lastUpdateTime))detaTime :&＃61; (endTime - startTime) / awardBatch.totalAmount// 计算下一个奖品释放的时间点releaseTime :&＃61; startTime &＃43; (totalAmount - totalBalance) * detaTime &＃43; int64(random.Int()) % detaTimelog.Println("relaseTime : ", time.Unix(releaseTime, 0).Format("2006-01-02 15:04:05"))if time.Now().Unix()

以上就是三种实现抽奖算法的思路&＃xff0c;以及抽奖算法的 Go 实现。对比思路一、二、三&＃xff0c;最终我们决定使用思路三进行实现。在思路上中&＃xff0c;牵涉到对 Redis 的操作&＃xff0c;所以这里我们需要先了解使用 Go 如何进行 Redis 操作。

Go 操作 Redis