礼物连发实现——请求合并

2024-03-10 07:20
文章标签 实现 请求 合并 礼物 连发

本文主要是介绍礼物连发实现——请求合并，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

一、业务实现

在全民直播的大环境下，大家对送礼业务也不再陌生，简单介绍一下一般情况下送礼业务的实现。

在这里插入图片描述

  • 1、通过Redis支持的lua脚本或watch命令，实现余额检查、预扣费两部操作，保证事务性，实现CAS操作；
  • 2、使用消息队列异步处理具体业务，提高业务的高并发能力，避免因为数据库连接数量限制了接收请求的上限；
  • 3、借助数据库的事务能力，保证送礼方、收礼方账户操作的事务性；
  • 4、常见消息推送服务均有发送频率限制，推荐由客户端发起请求，避免触频；

二、连发送礼问题

区别普通送礼连发送礼
场景用户单击，一次送礼用户不停的连击，多次送礼
并发情况大量的请求来自不同的用户同一个用户发出大量的请求
扩展能力分布式Redis增加分片，提高并发处理能力暂无
  • 从业务实现上我们可以看出，整个服务器接收能力的瓶颈在Redis上；
  • 对于普通送礼，我门通过使用分布式Redis，使不同用户可以均匀的分布在不同的Redis分片上，提高并发处理能力，后期也可以通过增加分片保证一定的扩展性；
  • 对于连发送礼的场景，同一个用户的请求均在同一个Redis分片，只能串行执行，导致整体Redis的冷热不均匀，出现某一个分片的CPU暴增，引起服务异常。
    在这里插入图片描述
    出现上述现象的核心原因是针对预扣费逻辑，使用了Redis的sortedset存储还未消费完成的事件，用户的余额检查通过用户实际的余额-未处理完的事件，当消费处理能力跟不上时，就会导致这个sortedset成为一个大key，后续对这个key的操作越来越慢，CPU持续上涨。

三、解决办法——请求合并

对于这种无法并发处理的业务，就需要采取降低事件的产生速度，此时就可以引入请求合并的思路，将多个相同的事件合并成一个事件。
当用户对一个礼物连击十次，正常会产生10个事件，那么就可以将其合成，只发送一个10个礼物的事件。

1、核心问题

要想实现连发请求的合并，我认为有以下几个问题需要解决

a、如何保证合并的请求均可以成功发送（余额足够）？

预测：通常情况下，用户在发起连发的过程中，账户不会出现其他的扣费行为，那么当用户发起第一次连发请求时，通过当前余额，可以预测用户最大可连发数量。

当然为了减少出现问题的可能性，可以保留一定的缓冲空间，达到阈值后的请求，均不在等待合并，直接发送同步执行。

b、由谁来进行请求合并？

由客户端进行合并业务会比较简单，但是对于已经上线的业务，为了避免发版，只能采取服务端合并。

c、何时将合并的请求发送？
  • 合并一定范围时间内的请求，比如1s内的连击事件；
  • 设定一次收集的阈值，具体值需要实践调整
d、合并后的请求是否需要阻塞等待处理结果？

不需要，在有下一个合并的请求出现后，上一个请求就可以结束掉，根据缓存信息构造返回值，避免出现大量的请求阻塞造成性能问题。

2、单机环境下实现

在这里插入图片描述

  • 使用LinkedBlockQueue阻塞队列，保存所有已合并的请求；
  • 使用LinkedBlockQueue阻塞队列，保存请求的返回值，合并后的请求阻塞获取结果，实现timeout机制；
  • 使用ReentrantLock保证并发操作的正确性。
private static final Cache<String, SeriesSendCacheInfo> QUEUE_CACHE = Caffeine.newBuilder().maximumSize(5)// 缓存 10min.expireAfterWrite(600, TimeUnit.SECONDS).build();@Datapublic static class SeriesSendCacheInfo {/*** 待处理请求*/private Queue<Queue<Object>> queue;/*** 预测连发信息*/private SeriesSendInfo4Redis seriesSendInfo4Redis;/*** 已合并数量*/private AtomicInteger mergeAmount;/*** 上一次请求*/private Queue<Object> lastRequest;/*** 同步锁*/private ReentrantLock lock;}public SuishenActionResult<SendGiftResultVo> send(SendGiftRequest sendGiftRequest) {if (StringUtils.isEmpty(sendGiftRequest.getSeriesSendId())) {return tipFlowService.sendGiftOperate(sendGiftRequest);}// 查询连发信息SeriesSendCacheInfo info = QUEUE_CACHE.get(sendGiftRequest.getSeriesSendId(), key -> {SeriesSendCacheInfo seriesSendCacheInfo = new SeriesSendCacheInfo();seriesSendCacheInfo.setQueue(new LinkedBlockingQueue<>());seriesSendCacheInfo.setSeriesSendInfo4Redis(giftSeriesSendInfoDao4Redis.getSeriesSendInfo(sendGiftRequest.getSeriesSendId()));seriesSendCacheInfo.setMergeAmount(new AtomicInteger());seriesSendCacheInfo.setLock(new ReentrantLock());return seriesSendCacheInfo;});SeriesSendInfo4Redis seriesSendInfo = info.getSeriesSendInfo4Redis();// 可聚合次数太少，同步处理if (Objects.isNull(seriesSendInfo.getSeriesSeconds()) || seriesSendInfo.getMergeTimes() <= 20) {return tipFlowService.sendGiftOperate(sendGiftRequest);}int mergeTimes = 0;Queue<Object> lastRequest = null;boolean pushSuccess = true;LinkedBlockingQueue<Object> ownQueue = null;ReentrantLock lock = info.getLock();lock.lock();try {if (info.getMergeAmount().get() >= seriesSendInfo.getMergeTimes()) {// 已达到最大合并数量pushSuccess = false;} else {// 更新已合并数量mergeTimes = info.getMergeAmount().incrementAndGet();// 获取上一次请求lastRequest = info.getLastRequest();ownQueue = new LinkedBlockingQueue<>();// 更新最后一次请求info.setLastRequest(ownQueue);// 加入待处理队列info.getQueue().add(ownQueue);}} finally {lock.unlock();}if (!pushSuccess) {// 未成功收集，优先消费已收集请求consume(sendGiftRequest);// 消费自己return tipFlowService.sendGiftOperate(sendGiftRequest);}if (Objects.nonNull(lastRequest)) {// 将上一次请求成功返回lastRequest.add(buildResponse(sendGiftRequest, seriesSendInfo));}// 已成功聚合if (info.getQueue().size() >= 10 || mergeTimes == seriesSendInfo.getMergeTimes()) {// 已聚合长度过长，开始消费SuishenActionResult<SendGiftResultVo> consume = consume(sendGiftRequest);if (Objects.nonNull(consume)) {// 自己已消费，返回消费结果return consume;}}// 自己未消费，阻塞等待消费结果Object result;try {result = ownQueue.poll(1, TimeUnit.SECONDS);} catch (InterruptedException e) {// 超时未获取到结果,自己主动开始消费result = consume(sendGiftRequest);}return result;}public SuishenActionResult<SendGiftResultVo> consume(SendGiftRequest sendGiftRequest, Queue<Queue<Object>> queue) {SuishenActionResult<SendGiftResultVo> result = null;List<Queue<Object>> requests = Lists.newArrayList();while (true) {// 获取待处理请求Queue<Object> request = queue.poll();if (Objects.isNull(request)) {break;}requests.add(request);}if (CollectionUtils.isEmpty(requests)) {return result;}// 更新请求参数sendGiftRequest.setOnceGiftNum(sendGiftRequest.getGiftNum());sendGiftRequest.setMergeTimes(requests.size());sendGiftRequest.setGiftNum(Math.multiplyExact(requests.size(), sendGiftRequest.getGiftNum()));// 发送请求SuishenActionResult<SendGiftResultVo> sendGiftResultVoSuishenActionResult = tipFlowService.sendGiftOperate(sendGiftRequest);requests.forEach(request -> {// 将结果放入结果队列request.add(sendGiftResultVoSuishenActionResult);});return result;}

3、分布式环境下实现

通过Nginx配置负载均衡策略，将同一个用户的请求分发到一台服务器上，使用上述单机环境的实现。这个方式可能会造成不同机器上的请求分配不均，所以介绍第二种方案，整体思路与单机环境大体相同。

  • 使用redis list结构代替单机方案中的LinkedBlockQueue阻塞队列；
  • 使用redis lua代替单机方案中的ReentrantLock，保证并发安全性
public SuishenActionResult<SendGiftResultVo> send(SendGiftRequest sendGiftRequest) {if (StringUtils.isEmpty(sendGiftRequest.getSeriesSendId())) {return tipFlowService.sendGiftOperate(sendGiftRequest);}// 查询信息SeriesSendInfo4Redis seriesSendInfo = giftSeriesSendInfoDao4Redis.getSeriesSendInfo(sendGiftRequest.getSeriesSendId());// 可聚合次数太少if (Objects.isNull(seriesSendInfo.getSeriesSeconds()) || seriesSendInfo.getMergeTimes() <= 20) {return tipFlowService.sendGiftOperate(sendGiftRequest);}String requestId = UUID.randomUUID().toString().replace("-", "");// 使用redis lua脚本实现，获取聚合结果SeriesPushResponse pushResponse = giftSeriesSendInfoDao4Redis.queuePush(sendGiftRequest.getSeriesSendId(),requestId, seriesSendInfo.getMergeTimes());if (pushResponse.pushSuccess()) {if (StringUtils.isNotEmpty(pushResponse.getLastRequest())) {// 将之前的请求结束giftSeriesSendInfoDao4Redis.queueResult(pushResponse.getLastRequest());}if (pushResponse.getMergeTimes() == 1) {// 缓存请求信息CompletableFuture.runAsync(() -> giftSeriesSendInfoDao4Redis.setSeriesRequestInfo(sendGiftRequest),myExecutor);}// 已成功聚合if (pushResponse.getQueueNum() >= 10 || pushResponse.getMergeTimes() == seriesSendInfo.getMergeTimes()) {// 已聚合长度过长，开始消费instance.appendLog("已聚合长度过长，开始消费");SuishenActionResult<SendGiftResultVo> consume = consume(sendGiftRequest, requestId);if (Objects.nonNull(consume)) {// 自己已消费，返回消费结果instance.appendLog("自己已消费，返回消费结果");return consume;}}} else {// 未成功收集，优先消费已收集请求instance.appendLog("未成功收集，优先消费已收集请求");consume(sendGiftRequest, requestId);// 消费自己instance.appendLog("消费自己");return tipFlowService.sendGiftOperate(sendGiftRequest);}// 自己未消费，阻塞等待消费结果String result = giftSeriesSendInfoDao4Redis.getResult(requestId, 1);if (Objects.isNull(result)) {// 超时未获取到结果,自己消费instance.appendLog("阻塞获取结果失败,自己消费");SuishenActionResult<SendGiftResultVo> consume = consume(sendGiftRequest, requestId);if (Objects.nonNull(consume)) {// 自己已消费，返回消费结果return consume;}// 自己未消费，被其它线程消费return buildResponse(sendGiftRequest, seriesSendInfo, pushResponse);}instance.appendLog("阻塞获取结果成功");return buildResponse(sendGiftRequest, seriesSendInfo, pushResponse);}public SuishenActionResult<SendGiftResultVo> consume(SendGiftRequest sendGiftRequest, String requestId) {LogContext instance = LogContext.instance();SuishenActionResult<SendGiftResultVo> result = null;// 使用redis lua获取队列请求List<String> requests = giftSeriesSendInfoDao4Redis.queueTake(sendGiftRequest.getSeriesSendId());if (CollectionUtils.isEmpty(requests)) {return result;}instance.appendLog("合成请求个数%s", requests.size());sendGiftRequest.setOnceGiftNum(sendGiftRequest.getGiftNum());sendGiftRequest.setMergeTimes(requests.size());sendGiftRequest.setGiftNum(Math.multiplyExact(requests.size(), sendGiftRequest.getGiftNum()));// 发送请求SuishenActionResult<SendGiftResultVo> sendGiftResultVoSuishenActionResult = tipFlowService.sendGiftOperate(sendGiftRequest);if (StringUtils.isNotEmpty(requestId) && requests.contains(requestId)) {result = sendGiftResultVoSuishenActionResult;}//             // 将结果放入结果队列，返回结果giftSeriesSendInfoDao4Redis.queueResult(requests.get(requests.size() - 1));return result;}private static volatile String pushEvalSha;private static volatile String popEvalSha;@PostConstructpublic void init() {new SuishenRedisExecutor<Boolean>().exe(jedis -> {// 初始化lua脚本pushEvalSha = jedis.scriptLoad(QUEUE_PUSH_LUA_SCRIPT);popEvalSha = jedis.scriptLoad(QUEUE_POP_LUA_SCRIPT);return true;}, msRedisTemplate);}public SeriesPushResponse queuePush(String sid, String requestId, int amount) {List<Object> list = new SuishenRedisExecutor<List<Object>>().exe(jedis ->(List<Object>) jedis.evalsha(pushEvalSha, Arrays.asList(RedisKeysOfTip.seriesMergeTimes(sid), RedisKeysOfTip.seriesQueue(sid)),Arrays.asList(String.valueOf(amount), requestId)), msRedisTemplate);return new SeriesPushResponse().setMergeTimes(NumberUtils.toInt(list.get(0).toString())).setLastRequest(list.size() > 2 ? list.get(2).toString() : null).setQueueNum(list.size() > 1 ? NumberUtils.toInt(list.get(1).toString()) : 0);}public List<String> queueTake(String sid) {return new SuishenRedisExecutor<List<String>>().exe(jedis ->(List<String>) jedis.evalsha(popEvalSha, 1, RedisKeysOfTip.seriesQueue(sid)), msRedisTemplate);}public void queueResult(String request) {distributeRedisTemplate.rpush(RedisKeysOfTip.seriesQueueResult(request), "1");}/*** 获取结果** @param request 请求Id* @param time    超时时间* @return 结果*/public String getResult(String request, int time) {List<String> brpop = distributeRedisTemplate.blpop(time, RedisKeysOfTip.seriesQueueResult(request));if (CollectionUtils.isNotEmpty(brpop)) {return brpop.get(0);}return null;}private static final String QUEUE_POP_LUA_SCRIPT ="local array = {}; \n" +"while( true )\n" +"do\n" +"   local req = redis.call(\"lpop\",KEYS[1]);\n" +"   if(req)\n" +"   then\n" +"       table.insert(array,req);\n" +"   else\n" +"       break;" +"   end\n" +"end\n" +"return array";private static final String QUEUE_PUSH_LUA_SCRIPT ="local array = {}; \n" +"local times = redis.call(\"get\",KEYS[1]);\n" +"if( times and tonumber(times) >= tonumber(ARGV[1]) )\n" +"then\n" +"   table.insert(array,times);\n" +"   return array;\n" +"end\n" +"times = redis.call(\"incr\",KEYS[1]);\n" +"table.insert(array,times);\n" +"local queue = tonumber(redis.call(\"rpush\",KEYS[2],ARGV[2]));\n" +"table.insert(array,queue);\n" +"if( queue > 1)\n" +"then\n" +"   table.insert(array,redis.call(\"LINDEX\",KEYS[2],queue-2));\n" +"end\n" +"return array;";

四、修改后性能对比

使用jmeter，模拟一个用户，连发3000次礼物，每次间隔100ms

原方案结果

  • 响应时长
    在这里插入图片描述

  • redis CPU监控
    在这里插入图片描述

合并后结果

  • 响应时长

在这里插入图片描述

  • redis CPU监控
    在这里插入图片描述

这篇关于礼物连发实现——请求合并的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/793474。

相关文章

Python实现精准提取 PDF中的文本,表格与图片

Python实现精准提取 PDF中的文本,表格与图片

《Python实现精准提取PDF中的文本,表格与图片》在实际的系统开发中,处理PDF文件不仅限于读取整页文本,还有提取文档中的表格数据,图片或特定区域的内容,下面我们来看看如何使用Python实... 目录安装 python 库提取 PDF 文本内容：获取整页文本与指定区域内容获取页面上的所有文本内容获取
阅读更多...
基于Python实现一个Windows Tree命令工具

基于Python实现一个Windows Tree命令工具

《基于Python实现一个WindowsTree命令工具》今天想要在Windows平台的CMD命令终端窗口中使用像Linux下的tree命令,打印一下目录结构层级树,然而还真有tree命令,但是发现... 目录引言实现代码使用说明可用选项示例用法功能特点添加到环境变量方法一：创建批处理文件并添加到PATH1
阅读更多...
Java使用HttpClient实现图片下载与本地保存功能

Java使用HttpClient实现图片下载与本地保存功能

《Java使用HttpClient实现图片下载与本地保存功能》在当今数字化时代,网络资源的获取与处理已成为软件开发中的常见需求,其中,图片作为网络上最常见的资源之一,其下载与保存功能在许多应用场景中都... 目录引言一、Apache HttpClient简介二、技术栈与环境准备三、实现图片下载与保存功能1.
阅读更多...
canal实现mysql数据同步的详细过程

canal实现mysql数据同步的详细过程

《canal实现mysql数据同步的详细过程》：本文主要介绍canal实现mysql数据同步的详细过程,本文通过实例图文相结合给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的... 目录1、canal下载2、mysql同步用户创建和授权3、canal admin安装和启动4、canal
阅读更多...
Nexus安装和启动的实现教程

Nexus安装和启动的实现教程

《Nexus安装和启动的实现教程》：本文主要介绍Nexus安装和启动的实现教程,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录一、Nexus下载二、Nexus安装和启动三、关闭Nexus总结一、Nexus下载官方下载链接：DownloadWindows系统根
阅读更多...
SpringBoot集成LiteFlow实现轻量级工作流引擎的详细过程

SpringBoot集成LiteFlow实现轻量级工作流引擎的详细过程

《SpringBoot集成LiteFlow实现轻量级工作流引擎的详细过程》LiteFlow是一款专注于逻辑驱动流程编排的轻量级框架,它以组件化方式快速构建和执行业务流程,有效解耦复杂业务逻辑,下面给大... 目录一、基础概念1.1 组件（Component）1.2 规则（Rule）1.3 上下文（Conte
阅读更多...
MySQL 横向衍生表(Lateral Derived Tables)的实现

MySQL 横向衍生表(Lateral Derived Tables)的实现

《MySQL横向衍生表(LateralDerivedTables)的实现》横向衍生表适用于在需要通过子查询获取中间结果集的场景,相对于普通衍生表,横向衍生表可以引用在其之前出现过的表名,本文就来... 目录一、横向衍生表用法示例1.1 用法示例1.2 使用建议前面我们介绍过mysql中的衍生表（From子句
阅读更多...
Mybatis的分页实现方式

Mybatis的分页实现方式

《Mybatis的分页实现方式》MyBatis的分页实现方式主要有以下几种,每种方式适用于不同的场景,且在性能、灵活性和代码侵入性上有所差异,对Mybatis的分页实现方式感兴趣的朋友一起看看吧... 目录​1. 原生 SQL 分页（物理分页）​​2. RowBounds 分页（逻辑分页）​​3. Page
阅读更多...
Python基于微信OCR引擎实现高效图片文字识别

Python基于微信OCR引擎实现高效图片文字识别

《Python基于微信OCR引擎实现高效图片文字识别》这篇文章主要为大家详细介绍了一款基于微信OCR引擎的图片文字识别桌面应用开发全过程,可以实现从图片拖拽识别到文字提取,感兴趣的小伙伴可以跟随小编一... 目录一、项目概述1.1 开发背景1.2 技术选型1.3 核心优势二、功能详解2.1 核心功能模块2.
阅读更多...
MYSQL查询结果实现发送给客户端

MYSQL查询结果实现发送给客户端

《MYSQL查询结果实现发送给客户端》：本文主要介绍MYSQL查询结果实现发送给客户端方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录mysql取数据和发数据的流程（边读边发）Sending to clientSending DataLRU(Least Rec
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2