红包雨-负责文档汇总

体验了一下字节的后端训练营，和小伙伴一起实现了一个红包雨系统的核心接口。这里记录一下我负责的部分文档，包括红包分配方案和流量削峰方案。

项目背景

春节红包雨是每年字节跳动最具挑战性的活动项目之一（除夕集中时间海量用户的参与）

需求说明

语言不限，实现 3 个接口：抢红包接口、拆红包接口、钱包列表接口

1. 抢红包接口
   • 每个用户最多只能抢到 N 次，次数可配
   • 一定概率能抢到，概率可配置
2. 拆红包接口
   • 把红包拆开入账到钱包

3. 钱包列表接口

   • 显示总余额

   • 显示抢到的红包，有已拆红包和未拆红包

   • 按红包获取时间排序

红包雨的总金额、总个数、每个红包的金额范围可配置(配置文件)

红包分配方案

红包的生成与分配是项目中非常重要的一环。一个好的方案应该能满足以下需求：

1. 尽可能把预算花完
2. 保证不超过预算
3. 当配置项临时变更时，仍应满足前三点需求

我们精心设计了一套红包生成与分配的方案，若流量充足，可以使得配置的红包数分发完毕时，我们的方案可以使得配置的总金额正好消耗完毕。同时，我们还设计了“锦鲤”机制，可以让部分用户抽到大额的锦鲤红包，增加活动的趣味性。

可配置项

• 普通红包总金额 $M$（保存为分）
• 普通红包个数 $N$
• 金额范围 $[min, max]$
• 用户最多可抢次数 $k$
• 抢到概率 $p$
• 锦鲤红包个数 $F$
• 锦鲤红包金额 $m_F$

由于引入了锦鲤红包，因此在我们的配置中需要对原始配置进行转换。设原始配置的红包总金额为$M_0$，红包总个数为$N_0$则转换公式为：

$$M_0=M+F*m_F$$ $$N_0=N+F$$

我们为配置内容引入了合理的约束：

$$min<=M/(N-F)<=max$$

该约束非常好理解，假定普通红包均值小于设定的最小值，或大于设定的最大值，说明此时的配置并不合理。在红包金额限定在$[min,max]$时，前者无法保证配置的总红包数消耗完，后者无法保证配置的总金额消耗完。

方案实现

对于一次 抢红包接口（/snatch） 调用，我们将其分为三个阶段完成。第一个阶段检查用户当前抢的次数，第二个阶段以配置的概率为依据判断是否抢到，第三个阶段为判定抢红包成功的请求分配红包实例，包括红包id（eid）和红包金额

阶段一：抢红包次数检查

该阶段通过检查用户抢的次数，可以让一批无效的请求不进入后续的计算环节，减轻服务压力的同时还可以提高响应速度。

流程：

用$S_{count}$表示调用用户当前抢的次数（用户第一次调用 /snatch 时设置为1）

检查$S_{count} \le k$是否满足：

• 若满足，令 $S_{count} = S_{count} + 1$，进入第二阶段
• 若不满足，直接返回抢红包失败

阶段二：判断是否抢到

该阶段通过对$p$做预处理，判定某一次成功的调用（即通过阶段一）是否抢到红包。

流程：

令$a / b = p / 1$，其中$a，b$均为整数，且$a / b$不可再约分

维护一个全局的请求计数器 $C$，计算 $C\ mod\ b$

• 若 $C\ mod\ b < a$，判定本次抢红包成功，进入阶段三
• 若 $C\ mod\ b \ge a$，判定本次抢红包失败

计算完成后，令 $C=C+1$

我们的方案规避了随机数库的不可控性，当流量足够大时，可以严格保证每次成功的调用抢到红包的概率是$p$，同时实现也相对简单。

核心代码：

func gcd(x, y int64) int64 {
   if x%y == 0 {
      return y
   }
   return gcd(y, x%y)
}

func loadStrategy(p float64) (int64, int64) {
   var b int64 = 1
   for p != float64(int64(p)) {
      b *= 10
      p *= 10
   }
   a := int64(p)
   d := gcd(b, a)
   return a / d, b / d
}

func testSnatching() bool {
   next := atomic.AddInt64(&C, 1)
   cur := next - 1
   return cur%a < b
}

在实际的编码中，我们采用原子操作来进行请求计数器$C$的自增，无需加锁的开销。

阶段三：红包金额分配

该阶段为成功抢到的红包分配红包id（表示为 eid ）和红包金额。

红包id生成

由于我们的应用最终要进行多实例部署，如何在分布式环境中保证 eid 的全局唯一性就成了这一阶段的重点。最终，我们的实现方案是 Redis分布式锁 + 分段申请，最终生成的 eid 范围为$[1,N+F]$

Redis中会维护一条剩余红包数的记录，在应用启动时，Redis会从配置文件中读取总红包个数$N+F$，并将其设置到剩余红包数的记录中。

每个实例启动后调用eid申请函数，其首先会获取一个由Redis实现的分布式锁，之后读取Redis中的剩余红包数$R$，则可以根据总红包数计算已分配的红包数$A=N-R$，此时下一个可用的 eid 即为$A+1$。之后，申请函数会尝试申请一段长度为$CacheCapacitiy$的 eid ，并设置$R=R-CacheCapacity$，最终本次申请到的 eid 范围为$[A+1,A+CacheCapacity]$，两个边界将被设置到实例本地。若剩余红包数$R<CacheCapacitiy$，则直接申请到边界，即最终申请到的 eid 范围为$[A+1,R]$，并设置$R=0$。后续的实例再进行申请时发现已没有剩余红包，则会设置本地的已发完标记，后续的 /snatch 调用都不再成功。

分段申请的设计可以减少分布式锁的加锁次数，提高系统性能。

每个实例还会在本地维护一个$offset$变量，用于标记下一个使用的 eid 相较于起始 eid 的偏移量。当请求达到时，每个实例会更新$offset$，并利用其分配 eid。

锦鲤红包的判定

在我们的方案中，锦鲤红包的个数和金额都是固定的，且锦鲤红包个数只占所有红包中很小一部分，因此我们的方案是在$[1,N]$中设置$F$个等分点，等分点对应的 eid 则判定为锦鲤红包。由于最终的请求流量是随机的，因此抢到锦鲤红包的用户也是随机的。

红包金额生成

在完成 eid 申请后，金额生成算法会为所有 eid 生成一个满足配置的金额。金额要么落在$[min,max]$中，要么为$m_F$。

Redis中会维护一条普通红包金额均值的记录，与剩余红包数类似，同样在应用启动时写入。

每次 eid 申请完成后，按如下流程为每个 eid 分配红包金额：

1. 读取Redis中的普通红包金额均值$Avg$

2. 每次生成一对红包，金额分别为$Avg-i, Avg+i$，其中$i \in N$且$Avg-i \ge min, Avg+i \le max$

   如果 eid 对应的是普通红包，则按以上规则生成；如果一对红包中有一个 eid 命中了锦鲤红包的 eid，则为其分配金额 $m_F$，同时为另一个红包分配金额$Avg$

3. 循环多次，直到最终为所有申请到的 eid 分配好金额。如果最后剩下一个红包，则根据其是否为锦鲤红包 eid 为其分配金额$m_F$或$Avg$

除锦鲤红包外，该方案可以保证每一次的原子生成（即生成一对或一个）都不会让剩余普通红包的均值发生变化，因此可以保证当红包数消耗完时，红包总金额也正好消耗完。

核心代码：

type FunctionConfiguration struct {
   Times       int32
   Probability float64
   Amount      int32
   Envelope    int32
   Min         int32
   Max         int32
   LuckyCount  int32
   LuckyMoney  int32
}

type SnatchService struct {
   cfg                         FunctionConfiguration
   runOut                      bool
   eidStart, eidEnd, eidOffset int32
   valueCache                  []int32
   valueLock                   sync.Mutex
}

// 缓存创建
func (s *SnatchService) createValueCache() {
   identifier, _ := redisService.DLock(dLockKey)
   remaining, _ := redisService.GetRemainEnvelopeAmount()

   // eid starts from 1
   s.eidStart = s.cfg.Envelope - remaining + 1

   // No available eid
   if s.eidStart > s.cfg.Envelope {
      log.Info("Run out of envelopes.")
      s.runOut = true
      _, _ = redisService.DUnlock(identifier, dLockKey)
      return
   }

   // Calculate eid end of this allocation
   s.eidEnd = s.eidStart + cacheCapacity - 1

   // Not enough eid
   if s.eidEnd > s.cfg.Envelope {
      s.eidEnd = s.cfg.Envelope
   }
   N := s.eidEnd - s.eidStart + 1

   // Update remaining
   _ = redisService.SetRemainEnvelopeAmount(remaining - N)

   // Read current average value
   avg, _ := redisService.GetAvgValue()
   _, _ = redisService.DUnlock(identifier, dLockKey)

   if N > 0 {
      log.Info(fmt.Sprintf("Get eid range [%v, %v]\n", s.eidStart, s.eidEnd))
   }

   s.eidOffset = 0
   
   // Generate value of envelopes
   for idx := int32(0); idx < N; {
      for i := int32(0); idx < N && avg-i >= s.cfg.Min && avg+i <= s.cfg.Max; i++ {
         if idx+1 < N {
            // Eid <- [1, Amount], Mod <- [0, Amount)
            if (s.eidStart+idx-1)%s.luckyRound != 0 && (s.eidStart+idx)%s.luckyRound != 0 || s.eidStart+idx-1 == 0 {
               s.valueCache[idx] = avg - i
               s.valueCache[idx+1] = avg + i
            } else {
               // If lucky eid is met, two red envelopes with value luckyValue and avgValue will be created
               var lucky, another int32
               if (s.eidStart+idx)%s.luckyRound == 0 {
                  lucky = idx
                  another = idx + 1
               } else {
                  lucky = idx + 1
                  another = idx
               }
               s.valueCache[lucky] = s.cfg.LuckyMoney
               s.valueCache[another] = avg
            }
            idx += 2
         } else if idx < N {
            if (s.eidStart+idx)%s.luckyRound != 0 {
               s.valueCache[idx] = avg
            } else {
               s.valueCache[idx] = s.cfg.LuckyMoney
            }
         }
      }
   }
}

// 获得下一个红包的eid和金额
func (s *SnatchService) nextRedEnvelope() (int32, int32) {
   s.valueLock.Lock()
   defer s.valueLock.Unlock()

   if s.eidOffset > s.eidEnd-s.eidStart {
      s.createValueCache()
   }

   if s.runOut {
      return -1, -1
   }

   eid, value := s.eidStart+s.eidOffset, s.valueCache[s.eidOffset]
   s.eidOffset++
   return eid, value
}

应对配置更新

配置更新时需要完成两件事：一是更新Redis中的共享变量（剩余红包数、普通红包均值），二是更新各个实例的本地变量。

Redis更新

在多实例环境下，Redis更新的难点在于需要实现一种“仅写一次”的机制，即仅由一个实例完成写，但所有实例都需要读。

我们通过一种基于版本的方式来实现这一机制。我们引入了 Version 字段，用于描述配置的版本。每次更新时填入的版本号自增。Version 字段同时存在于Redis和配置文件中。和剩余红包数等共享变量类似，Version 会在应用初始化时从配置文件读入到Redis中。

每次触发配置更新的回调时，所有实例首先读取Redis中的 Version 字段，比较其和新配置文件中 Version 字段的大小。若前者等于后者，说明Redis中的配置已经被修改，当前实例无需再次写入。若前者小于后者，说明此时Redis中的配置尚未被修改，当前实例将尝试获取分布式锁。实例获取到锁后，将再一次对二者进行比较（双检锁），若仍满足前者小于后者，将执行写Redis操作。写完毕后，更新Redis中的 Version 字段，并释放分布式锁。

写Redis实质就是依据新配置更新共享变量剩余红包数$R$和普通红包均值$Avg$。由于更新Redis的实例此时持有老版本配置，因此可以依据普通红包总数$N$计算出已发放的普通红包数$S$。由于我们的方案保证了金额分配中$Avg$的值不变，因此使用$Avg*S$就可以计算出已发普通红包的金额。此时，无论是已发放的红包数还是已发放的金额，都有可能大于等于新配置中的对应量。若检查到这种情况，说明此时在新配置下已经超发，Redis中的$R$和$Avg$都将被设为0。若二者都小于新配置，通过减去已发放的，我们可以计算出新的$R$和$Avg$，并将二者更新到Redis中。我们还对新配置增加了约束检查。当新的$Avg$小于新的$min$时，我们将设置$Avg=min$，即始终以$min$的面额发放红包，直到把预算消耗完（此时配置的红包数有剩余）。当新的$Avg$大于$max$时，我们将设置$Avg=max$，即始终以$max$的面额发放红包，直到把配置的红包数消耗完（此时预算有剩余）。

本地变量更新

每个实例在完成上一步后，需要依据新的配置对本地变量进行更新。这一步相对简单，只需读取Redis中的剩余红包数，检查新配置是否超发（为0），若超发，设置已发完标记，之后的发红包请求将直接失败。若未超发，将新配置中的变量覆盖本地变量即可。

由于Local Cache的存在，配置更新并不会立刻生效。当实例把消耗完Local Cache，就会按照新的配置进行红包申请了。

核心代码：

s := new(SnatchService)
func callback(cfg config.Configuration) {
   log.Info("Reloading function configuration...")

   // Keep this order of acquiring lock to avoid deadlock
   // valueLock -> DLock
   s.valueLock.Lock()
   defer s.valueLock.Unlock()

   oldCfg := s.cfg
   newCfg := cfg.Function
   var nowVersion int32

   // nowVersion == newCfg.Version means Redis has been updated,
   // there is no need to acquire lock again
   nowVersion, _ = redisService.GetFunctionVersion()
   if nowVersion < newCfg.Version {
      identifier, _ := redisService.DLock(dLockKey)
      nowVersion, _ = redisService.GetFunctionVersion()

      // Double check lock
      if nowVersion < newCfg.Version {
         remainingEnvelope, _ := redisService.GetRemainEnvelopeAmount()
         avg, _ := redisService.GetAvgValue()

         allocatedEnvelope := oldCfg.Envelope - remainingEnvelope
         allocatedAmount := allocatedEnvelope * avg

         newRemainingEnvelope := newCfg.Envelope - allocatedEnvelope
         newRemainingAmount := newCfg.Amount - allocatedAmount

         if newRemainingEnvelope <= 0 || newRemainingAmount <= 0 {
            // Allocated red envelopes have exceeded new configuration
            _ = redisService.SetRemainEnvelopeAmount(0)
            _ = redisService.SetAvgValue(0)
            log.Warn("Configuration mismatch: allocated red envelopes have exceeded new configuration.")
         } else {
            // Update average value for future local cache
            newAvg := newRemainingAmount / newRemainingEnvelope
            if newAvg < newCfg.Min {
               newAvg = newCfg.Min

               // Decrease the number of envelopes to keep the consistency
               // avg * remaining == remaining amount
               newRemainingEnvelope = newRemainingAmount / newCfg.Min
            } else if newAvg > newCfg.Max {
               newAvg = newCfg.Max
            }
            _ = redisService.SetRemainEnvelopeAmount(newRemainingEnvelope)
            _ = redisService.SetAvgValue(newAvg)
            log.Info(fmt.Sprintf("New configuration loaded, RemainingEnvelopes: %v , Avg: %v\n", newRemainingEnvelope, newAvg))
         }

         // Update configuration version in Redis
         _ = redisService.SetFunctionVersion(newCfg.Version)
      }

      _, _ = redisService.DUnlock(identifier, dLockKey)
   }

   // Update local variables
   s.updateConfiguration(newCfg)
}

func (s *SnatchService) updateConfiguration(cfg config.FunctionConfiguration) {
   remainingEnvelope, _ := redisService.GetRemainEnvelopeAmount()
   s.runOut = remainingEnvelope == 0
   if s.runOut {
      // Invalidate local cache
      s.eidOffset = 0
      s.eidEnd = -1
      s.eidStart = 0
   }
   s.loadConfiguration(cfg)
}

流量削峰方案

为了提高接口响应速度，每次数据的更新我们都会先将其写入速度更快的Redis，同时引入了消息队列来对数据库进行异步写入。考虑到应尽可能降低缓存与数据库不一致的时间，我们选择了消息时延更低的 RocketMQ 。

消息格式设计

每个用户的抢红包次数更新频繁，且时效性弱，通常只在一场活动中有效，我们选择将其完全保存在缓存中，不做入库处理。数据库中实际保存的只有用户钱包余额与红包信息。因此，我们设计了两种消息格式：

• insert：(eid int32, uid int32, value int32, snatchTime int64)

生成新红包时产生，用于更新数据库中的红包表。

• update: (eid int32, uid int32, value int32)

拆红包时产生，用于更新数据库中的用户余额表。

其中 insert 和 update 分别表示消息TAG，用于在消息消费端进行区分。

insert 消息记录了除 opened 字段外一条红包记录所需的所有信息，opened 字段无需传递，因为在红包创建的时刻其一定是未打开的。每一条 insert 消息在消费端被成功消费后都会在红包表中生成一条记录。

update 消息在用户拆红包时生成，消费端会根据 eid 更新红包的 opened 状态，同时修改用户余额。update 消息中同时携带了 uid 和 value ，可以减少一次红包表的查询。

消息顺序约束

一个红包只有在创建之后才能被拆，且只能拆一次。这样两个有效的操作首先会写Redis，之后将分别创建一条 insert 和一条 update 消息，并通过消息队列写往数据库。为了保证数据的一致性，我们需要确保 insert 消息一定要在 update 消息前被消费成功，否则就会出现在缓存中该红包已拆开，而在数据库中没有的情况。

RocketMQ提供了全局顺序消息的支持，但我们的业务并不需要保证消息的全局顺序性。RocketMQ还提供了一种更高效的分区顺序消息，即在发送消息时可以指定 Sharding Key ，具有相同 Sharding Key 的消息将被发往同一个物理队列。我们以 eid 作为 Sharding Key，可以保证有效的 update 消息一定会在对应的 insert 消息之后被消费。

Reference

项目总结文档