# start

[TOC]

# 1 直播间红包雨功能的设计与分析

1. **生成红包逻辑**

1. **红包数据的存储位置如何设计？**

- 临时性数据，持久化存在MySQL，内存当中，读写性能非常好。**通常做法是**：`本地内存`存储红包雨数据 或者 `redis`存储红包雨数据。 主播id -> 红包雨配置（唯一的code）

- 1.`使用Redis的List`集合存储红包数据（key包含code）-> pop接口领取红包（单个List存储1k个红包）（redis的分片存储，100个list）

- 缺点：每个List存储的数据量不能太大，否则有大key风险，但是可以利用 分片思路进行存储

- 2.`本地内存存储红包`数据。（500台tomcat，每台tomcat的本地内存里存储一定量的红包数据。20000个红包，一共是1000w个红包，2亿元红包，1kw人瓜分）（结合dns访问所在地最近的一台clb，clb-> nginx集群 ->gateway -> tomcat）

- 缺点：500台tomcat，假设其中的某一台挂了，导致这台tomcat的红包没被领完，但是对于平台来说并不亏

2. **红包的领取是怎么一个过程？**

- Redis + List存储红包数据（每场直播对应一个List，1000个红包）

- 如果按照主播id为维度去生成红包数据是怎么存储？

- List（一定要有一个唯一的标识，key最好有一个唯一的code识别）

- A主播开了一场直播，生成了一场红包雨的数据，但是没有领完，然后下播了。此时A主播下播后立马又获取了下一场直播的红包雨奖励。

- list:(id) -> (上一场没有领完的红包数据)

- list:(id) -> (往旧的list集合中塞入了红包数据)

3. **生成红包的流程是怎样的？**

- 主播点击前端的一个按钮去生成的。

- 如何防止红包重复生成？

- 设置红包雨配置信息的状态，并进行限流防止重复点击

4. **如何保证红包生成的均匀性？如何防止不超出预期金额限制？**

2. **抽红包逻辑**

1. 领取红包的流程如何设计？（redis的List的pop接口领取红包，如果是本地内存的话，queue也有pop功能）

2. 如何防止恶意刷接口调用领取红包？（A直播间，恶意请求可以领取到B，C，D等多个直播间的数据）

- 通过随机生成的红包雨code为标识进行抢红包，而不是以roomId或anchorId为标识

3. 如何通知全场观众开始参与抢红包行为？（`点击生成红包按钮，im异步通知主播，红包初始化完成，主播才有权限点击下一步，开始抢红包`）

4. 直播间之间的领取红包接口不能直接互通，如何考虑接口安全性？

> 总结一下整个红包雨开始的流程：

>

> 主播点击准备开始红包雨 --> 后台根据红包雨配置生成红包金额列表(存储到Redis或者本地内存) --> 通知主播数据准备好 --> 开始抢红包

# 2 红包雨功能的实现

## 2.1 红包雨配置对象基本操作的实现

在之前**qiyu-live-gift-provider**操作的数据库中新建红包雨配置表格：

```sql

-- Create syntax for TABLE 't_red_packet_config'

CREATE TABLE `t_red_packet_config` (

`id` int unsigned NOT NULL AUTO_INCREMENT,

`anchor_id` bigint NOT NULL DEFAULT '0' COMMENT '主播id',

`start_time` datetime DEFAULT NULL COMMENT '红包雨活动开始时间',

`total_get` int NOT NULL DEFAULT '0' COMMENT '一共领取数量',

`total_get_price` int NOT NULL DEFAULT '0' COMMENT '一共领取金额',

`max_get_price` int NOT NULL DEFAULT '0' COMMENT '最大领取金额',

`status` tinyint NOT NULL DEFAULT '1' COMMENT '(1 待准备，2已准备，3已发送)',

`total_price` int NOT NULL DEFAULT '0' COMMENT '红包雨总金额数',

`total_count` int unsigned NOT NULL DEFAULT '0' COMMENT '红包雨总红包数',

`config_code` varchar(250) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NOT NULL COMMENT '唯一code',

`remark` varchar(250) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NOT NULL DEFAULT '' COMMENT '备注',

`create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,

`update_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,

PRIMARY KEY (`id`)

) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='直播间红包雨配置';

INSERT INTO `qiyu_live_user`.`t_red_packet_config`

(`id`, `anchor_id`, `start_time`, `total_get`, `total_get_price`, `max_get_price`,

`status`, `total_price`, `total_count`, `config_code`, `remark`, `create_time`, `update_time`)

VALUES (3, 285608972927369200, NULL, 0, 0, 0, 1, 10000, 1000, '1', '默认红包雨配置',

'2024-02-24 22:28:47', '2024-02-24 22:28:47');

```

**qiyu-live-gift-provider：**

```java

@Data

@TableName("t_red_packet_config")

public class RedPacketConfigPO {

@TableId(type = IdType.AUTO)

private Integer id;

private Long anchorId;

private Date startTime;

private Integer totalGet;

private Integer totalGetPrice;

private Integer maxGetPrice;

private Integer status;

private Integer totalPrice;

private Integer totalCount;

private String configCode;

private String remark;

private Date createTime;

private Date updateTime;

}

```

```java

@Data

public class RedPacketConfigReqDTO implements Serializable {

@Serial

private static final long serialVersionUID = 5117539613836783248L;

private Integer id;

private Integer roomId;

private Integer status;

private Long userId;

private String redPacketConfigCode;

private Integer totalPrice;

private Integer totalCount;

private String remark;

}

```

```java

//gift-interface中

@Data

public class RedPacketConfigRespDTO implements Serializable {

@Serial

private static final long serialVersionUID = 5117539613836783248L;

private Long anchorId;

private Integer totalPrice;

private Integer totalCount;

private String configCode;

private String remark;

}

```

```java

//gift-interface中

@Data

@AllArgsConstructor

public class RedPacketReceiveDTO implements Serializable

```

```java

//gift-interface中

public enum RedPacketStatusEnum {

WAIT(1,"待准备"),

IS_PREPARED(2, "已准备"),

IS_SEND(3, "已发送");

int code;

String desc;

RedPacketStatusEnum(int code, String desc) {

this.code = code;

this.desc = desc;

}

public int getCode() {

return code;

}

public String getDesc() {

return desc;

}

}

```

```java

//gift-interface中

public interface IRedPacketConfigRpc {

/**

* 根据主播id查询有无发放红包雨的特权

*/

RedPacketConfigRespDTO queryByAnchorId(Long anchorId);

/**

* 新增红包雨配置

*/

boolean addOne(RedPacketConfigReqDTO redPacketConfigReqDTO);

/**

* 准备生成红包金额列表

*/

boolean prepareRedPacket(Long anchorId);

/**

* 直播间用户领取红包

*/

RedPacketReceiveDTO receiveRedPacket(RedPacketConfigReqDTO redPacketConfigReqDTO);

/**

* 开始红包雨

*/

Boolean startRedPacket(RedPacketConfigReqDTO reqDTO);

}

```

```java

@DubboService

public class RedPacketConfigRpcImpl implements IRedPacketConfigRpc {

@Resource

private IRedPacketConfigService redPacketConfigService;

@Override

public RedPacketConfigRespDTO queryByAnchorId(Long anchorId) {

return ConvertBeanUtils.convert(redPacketConfigService.queryByAnchorId(anchorId), RedPacketConfigRespDTO.class);

}

@Override

public boolean addOne(RedPacketConfigReqDTO redPacketConfigReqDTO) {

return redPacketConfigService.addOne(ConvertBeanUtils.convert(redPacketConfigReqDTO, RedPacketConfigPO.class));

}

@Override

public boolean prepareRedPacket(Long anchorId) {

return redPacketConfigService.prepareRedPacket(anchorId);

}

@Override

public RedPacketReceiveDTO receiveRedPacket(RedPacketConfigReqDTO redPacketConfigReqDTO) {

return redPacketConfigService.receiveRedPacket(redPacketConfigReqDTO);

}

@Override

public Boolean startRedPacket(RedPacketConfigReqDTO reqDTO) {

return redPacketConfigService.startRedPacket(reqDTO);

}

}

```

```java

public interface IRedPacketConfigService

@Override

public RedPacketConfigPO queryByConfigCode(String code)

@Override

public boolean addOne(RedPacketConfigPO redPacketConfigPO)

@Override

public boolean updateById(RedPacketConfigPO redPacketConfigPO) {

return redPacketConfigMapper.updateById(redPacketConfigPO) > 0;

}

@Override

public boolean prepareRedPacket(Long anchorId) {

return true;

}

@Override

public RedPacketReceiveDTO receiveRedPacket(String code) {

return null;

}

@Override

public Boolean startRedPacket(RedPacketConfigReqDTO reqDTO) {

return true;

}

}

```

## 2.2 红包雨金额列表的生成和存储

- **准备：**

redis的keyBuilder添加新的keyBuilder：

```java

@Configuration

@Conditional(RedisKeyLoadMatch.class)

public class GiftProviderCacheKeyBuilder extends RedisKeyBuilder

public String buildRedPacketInitLock(String code)

public String buildRedPacketTotalGetCount(String code)

public String buildRedPacketTotalGetPrice(String code)

public String buildRedPacketMaxGetPrice(String code)

public String buildUserTotalGetPrice(Long userId)

public String buildRedPacketPrepareSuccess(String code)

public String buildRedPacketNotify(String code)

}

```

写一个工具类ListUtils：可以将大列表拆分为多个子列表

```java

/**

* 将List拆分为小的List，用于Redis中list的存入，避免 Redis的输入输出缓冲区 和 网络 发生堵塞

*/

public class ListUtils else {

subList = list.subList(priIndex, list.size());

}

if (subList.size() > 0) {

resultList.add(subList);

}

}

return resultList;

}

}

```

- **开始书写逻辑：**

> - 我们要用到上面写的ListUtils中的**拆分列表**功能，用于将生成的红包金额列表拆分为小列表，再往redis中存，避免Redis的输入输出缓冲区堵塞

> - 我们生成红包金额列表 使用的算法是 **二倍均值法**

```java

@Service

public class RedPacketConfigServiceImpl implements IRedPacketConfigService

// 加锁保证原子性：仿重

Boolean isLock = redisTemplate.opsForValue().setIfAbsent(cacheKeyBuilder.buildRedPacketInitLock(redPacketConfigPO.getConfigCode()), 1, 3L, TimeUnit.SECONDS);

if (Boolean.FALSE.equals(isLock)) {

return false;

}

Integer totalPrice = redPacketConfigPO.getTotalPrice();

Integer totalCount = redPacketConfigPO.getTotalCount();

List priceList = this.createRedPacketPriceList(totalPrice, totalCount);

String cacheKey = cacheKeyBuilder.buildRedPacketList(redPacketConfigPO.getConfigCode());

// 将红包数据拆分为子集合进行插入到Redis，避免 Redis输入输出缓冲区 被填满

List> splitPriceList = ListUtils.splistList(priceList, 100);

for (List priceItemList : splitPriceList) {

redisTemplate.opsForList().leftPushAll(cacheKey, priceItemList.toArray());

}

// 更改红包雨配置状态，防止重发

redPacketConfigPO.setStatus(RedPacketStatusEnum.IS_PREPARED.getCode());

this.updateById(redPacketConfigPO);

// Redis中设置该红包雨已经准备好的标记

redisTemplate.opsForValue().set(cacheKeyBuilder.buildRedPacketPrepareSuccess(redPacketConfigPO.getConfigCode()), 1, 1L, TimeUnit.DAYS);

return true;

}

/**

* 二倍均值法：

* 创建红包雨的每个红包金额数据

*/

private List createRedPacketPriceList(Integer totalPrice, Integer totalCount)

int maxLimit = (totalPrice / (totalCount - i)) * 2;// 最大限额为平均值的两倍

int currentPrice = ThreadLocalRandom.current().nextInt(1, maxLimit);

totalPrice -= currentPrice;

redPacketPriceList.add(currentPrice);

}

return redPacketPriceList;

}

}

```

## 2.3 开始红包雨逻辑

> 主播接收到红包数据准备完毕后，就可以点击开始红包雨了

- **准备：**

```java

public enum ImMsgBizCodeEnum

// 红包已经开始过（有别的线程正在通知用户中），返回false

String notifySuccessCacheKey = cacheKeyBuilder.buildRedPacketNotify(code);

if (Boolean.TRUE.equals(redisTemplate.hasKey(notifySuccessCacheKey))) {

return false;

}

redisTemplate.opsForValue().set(notifySuccessCacheKey, 1, 1L, TimeUnit.DAYS);

// 广播通知直播间所有用户开始抢红包了

RedPacketConfigPO redPacketConfigPO = this.queryByConfigCode(code);

JSONObject jsonObject = new JSONObject();

jsonObject.put("redPacketConfig", JSON.toJSONString(redPacketConfigPO));

LivingRoomReqDTO livingRoomReqDTO = new LivingRoomReqDTO();

livingRoomReqDTO.setRoomId(reqDTO.getRoomId());

livingRoomReqDTO.setAppId(AppIdEnum.QIYU_LIVE_BIZ.getCode());

List userIdList = livingRoomRpc.queryUserIdsByRoomId(livingRoomReqDTO);

if (CollectionUtils.isEmpty(userIdList)) return false;

this.batchSendImMsg(userIdList, ImMsgBizCodeEnum.RED_PACKET_CONFIG.getCode(), jsonObject);

// 更改红包雨配置的状态为已发送

redPacketConfigPO.setStatus(RedPacketStatusEnum.IS_SEND.getCode());

this.updateById(redPacketConfigPO);

return true;

}

/**

* 批量发送im消息

*/

private void batchSendImMsg(List userIdList, Integer bizCode, JSONObject jsonObject) );

routerRpc.batchSendMsg(imMsgBodies);

}

}

```

## 2.4 领红包逻辑

```java

//gift-interface中

/**

* 用户红包雨抢红包后发送的mq消息体

*/

@Data

public class SendRedPacketBO implements Serializable {

@Serial

private static final long serialVersionUID = 1829802295999336708L;

private Integer price;

private RedPacketConfigReqDTO reqDTO;

}

```

```java

public class GiftProviderTopicNames {

...

/**

* 用户红包雨抢红包消息topic

*/

public static final String RECEIVE_RED_PACKET = "receive-red-packet";

}

```

```java

@Service

public class RedPacketConfigServiceImpl implements IRedPacketConfigService

Integer price = (Integer) priceObj;

// 发送mq消息进行异步信息的统计，以及用户余额的增加

SendRedPacketBO sendRedPacketBO = new SendRedPacketBO();

sendRedPacketBO.setPrice(price);

sendRedPacketBO.setReqDTO(redPacketConfigReqDTO);

CompletableFuture> sendResult = kafkaTemplate.send(GiftProviderTopicNames.RECEIVE_RED_PACKET, JSON.toJSONString(sendRedPacketBO));

try receive a redPacket, send success", redPacketConfigReqDTO.getUserId());

}

}).exceptionally(e -> {

LOGGER.error("[RedPacketConfigServiceImpl] send error, userId is {}, price is {}", redPacketConfigReqDTO.getUserId(), price);

throw new RuntimeException(e);

});

} catch (Exception e) {

return new RedPacketReceiveDTO(null, "抱歉，红包被人抢走了，再试试");

}

return new RedPacketReceiveDTO(price, "恭喜领取到红包：" + price + "旗鱼币！");

}

...

}

```

> 刚刚我们发送了mq消息去做数据库工作：

在gift-provider中编写一个consumer来消费刚刚的mq消息：

```java

/**

* 处理抢红包mq消息的消费者

*/

@Component

public class ReceiveRedPacketConsumer catch (Exception e) {

LOGGER.error("[ReceiveRedPacketConsumer] receiveRedPacket error, mqBody is {}", sendRedPacketBOStr);

}

}

}

```

里面调用了redPacketConfigService.receiveRedPacketHandler()，现在来进行编写该方法

```java

public interface IRedPacketConfigService {

...

/**

* 接收到抢红包的消息过后，进行异步处理的handler

*/

void receiveRedPacketHandler(RedPacketConfigReqDTO reqDTO, Integer price);

}

```

```java

@Service

public class RedPacketConfigServiceImpl implements IRedPacketConfigService

```

> - 这里我们的取出红包的逻辑，因为我们的totalGetCountCacheKey和 totalGetPriceCacheKey使用的是Hash类型，所以我们之后**红包雨结束后，记得remove掉code对应的键值对**

> - **可以看到我们在用户每次抢红包时，就把金额incr持久化到了MySQL**

> - 但是：前面在Redis中又累加了每个用户抢到的金额，但是却没有用到

> - 并且：前面在Redis中的totalGetCount和totalGetPrice都没有用到

> - 所以：我推荐：

> - 方法一：`最后不做那3行持久化到MySQL的操作，应该最后红包雨完成后，由前端发起一个红包雨结束的请求，然后使用mq异步的处理MySQl持久化`，包括 1、从Redis读取totalGetCount和totalGetPrice的进行MySQL持久化；2、从Redis读取每个用户的累加金额进行MySQL数据库的持久化；3、移除Redis中的数据威灵顿凤凰数据分析

> - 方法二：`不做Redis的操作`，避免掉冗余代码

## 2.5 API模块调用逻辑

**qiyu-live-api：**

- **准备工作：**

```java

@Data

public class LivingRoomInitVO {

private Long anchorId;

private Long userId;

private String nickName;

private String anchorImg;

private String roomName;

private boolean isAnchor;

private String redPacketConfigCode;

private String avatar;

private Integer roomId;

private String watcherNickName;

private String anchorNickName;

//观众头像

private String watcherAvatar;

//默认背景图，为了方便讲解使用

private String defaultBgImg;

private Long pkObjId;

}

```

```java

@Data

public class RedPacketReceiveVO {

private Integer price;

private String msg;

}

```

```java

@Data

public class LivingRoomReqVO {

private Integer type;

private int page;

private int pageSize;

private Integer roomId;

private String redPacketConfigCode;

}

```

- **开始逻辑的编写：**

> 设置好我们的@RequestLimit进行限流，防止重复点击

```java

@RestController

@RequestMapping("/living")

public class LivingRoomController

@RequestLimit(limit = 1, second = 10, msg = "正在广播直播间用户，请稍等")

@PostMapping("/startRedPacket")

public WebResponseVO startRedPacket(LivingRoomReqVO livingRoomReqVO)

@RequestLimit(limit = 1, second = 7, msg = "")

@PostMapping("/getRedPacket")

public WebResponseVO getRedPacket(LivingRoomReqVO livingRoomReqVO)

}

```

```java

public interface ILivingRoomService

```

> 下一段代码LivingRoomServiceImpl里会调用livingRoomRpc.queryByAnchorId(userId)，但是我们的living-provider里还没有写这个方法，我们需要使用在其中添加这个方法，LivingRoomServiceImpl的代码如下，RPC接口和service接口的代码请自己补充：

>

> ```java

> @Service

> public class LivingRoomServiceImpl implements ILivingRoomService

> ```

```java

@Service

public class LivingRoomServiceImpl implements ILivingRoomService

@Override

public Boolean startRedPacket(Long userId, String code)

@Override

public RedPacketReceiveVO getRedPacket(Long userId, String code) else

return respVO;

}

}

```

# end