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

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

一、业务实现

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

在这里插入图片描述

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;";