一、什么是请求合并

在WEB项目中，我们一般会使用HTTP协议来处理请求

那么我们与服务器交互方式将会是这样的，一次请求，一次处理

我们都知道，调用批量接口相比调用非批量接口有更大的性能优势（因为减少了IO交互操作），在高并发情况下，如果有非常频繁的接口请求发生的话，我们则可以考虑请求合并了，将多个请求进行一定的等待延迟，当请求累计达到一定量级的时候，进行批量请求处理

二、请求合并的优缺点

所谓请求合并，就是讲多次请求合并为一次批量请求

优点:

将多次请求处理进行一定时间或请求数量的等待，使之合并成为一次请求，减少IO交互

缺点:

由于请求需要等待指定时间或指定请求数量，所以合并的接口存在延时，故对请求合并的接口有所限制，该接口不能对响应及时性有要求，支持一定时间的延迟

三、请求合并技术实现

采用定时线程池ScheduledExecutorService，与内存队列LinkedBlockingDeque进行实现请求合并

原理是将用户的请求进行缓存起来，缓存的请求数量达到指定数量或达到定时线程池执行时，将已有多个单请求处理合并为多处理，调用批量接口进行操作

依赖

只需要JDK,无需任何第三方依赖

批量请求合并工具类定义如下:

核心原理就是将请求放入队列，放入时检测内存队列数量是否超过设置阈值，以及时间阈值到期触发定时线程池执行

package com.leilei.support;import lombok.extern.log4j.Log4j2;import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;/*** @author lei* @create 2022-03-23 14:39* @desc 请求合并工具类**/
@Log4j2
public class BatchCollapser<T, R> {private static final Map<Class, BatchCollapser> BATCH_INSTANCE =new ConcurrentHashMap<>();private static final ScheduledExecutorService SCHEDULE_EXECUTOR = Executors.newScheduledThreadPool(1);private final LinkedBlockingDeque<T> batchContainer = new LinkedBlockingDeque<>();private final BatchHandler<List<T>, R> handler;private final int countThreshold;/*** constructor** @param handler        处理器* @param countThreshold 数量阈值，达到此阈值后触发处理器* @param timeThreshold  时间阈值，达到此时间后触发处理器*/private BatchCollapser(BatchHandler<List<T>, R> handler, int countThreshold, long timeThreshold) {this.handler = handler;this.countThreshold = countThreshold;SCHEDULE_EXECUTOR.scheduleAtFixedRate(() -> {try {this.popUpAndHandler(BatchHandlerType.BATCH_HANDLER_TYPE_TIME);} catch (Exception e) {log.error("pop-up container exception", e);}}, timeThreshold, timeThreshold, TimeUnit.SECONDS);}/*** 添加请求元素入队* @param event*/public void addRequestParam(T event) {batchContainer.add(event);if (batchContainer.size() >= countThreshold) {popUpAndHandler(BatchHandlerType.BATCH_HANDLER_TYPE_DATA);}}/*** 从队列获取请求,并进行批量处理* @param handlerType*/private void popUpAndHandler(BatchHandlerType handlerType) {List<T> tryHandlerList = Collections.synchronizedList(new ArrayList<>(countThreshold));batchContainer.drainTo(tryHandlerList, countThreshold);if (tryHandlerList.size() < 1) {return;}try {R handle = handler.handle(tryHandlerList, handlerType);log.info("批处理工具执行result:{}", handle);} catch (Exception e) {log.error("batch execute error, transferList:{}", tryHandlerList, e);}}/*** 获取合并器实例** @param batchHandler   处理执行器* @param countThreshold 阈值数量(队列数量)* @param timeThreshold  阈值时间 单位秒（目前设置是触发后获取阈值数量请求，可根据需要修改）* @param <E>* @param <R>* @return*/public static <E, R> BatchCollapser<E, R> getInstance(BatchHandler<List<E>, R> batchHandler, int countThreshold, long timeThreshold) {Class jobClass = batchHandler.getClass();if (BATCH_INSTANCE.get(jobClass) == null) {synchronized (BatchCollapser.class) {BATCH_INSTANCE.putIfAbsent(jobClass, new BatchCollapser<>(batchHandler, countThreshold, timeThreshold));}}return BATCH_INSTANCE.get(jobClass);}/*** 请求处理接口** @param <T>* @param <R>*/public interface BatchHandler<T, R> {/*** 处理用户具体请求** @param input* @param handlerType* @return*/R handle(T input, BatchHandlerType handlerType);}/*** 合并执行类型枚举*/public enum BatchHandlerType {/*** 数量类型*/BATCH_HANDLER_TYPE_DATA,/*** 时间类型*/BATCH_HANDLER_TYPE_TIME,}
}

使用方式如下:

package com.leilei.support;import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;import javax.annotation.PostConstruct;
import java.util.List;/*** @author lei* @create 2022-03-22 15:22* @desc**/
@Service
public class ProductService implements BatchCollapser.BatchHandler<List<Integer>, Integer> {private BatchCollapser<Integer, Integer> batchCollapser;@PostConstructprivate void postConstructorInit() {// 当请求数量达到20个，或每过5s合并执行一次请求batchCollapser = BatchCollapser.getInstance(ProductService.this, 20, 5);}@Overridepublic Integer handle(List<Integer> input, BatchCollapser.BatchHandlerType handlerType) {System.out.println("处理类型:" + handlerType + ",接受到批量请求参数:" + input);return input.stream().mapToInt(x -> x).sum();}/*** 假设我这里是300ms一次请求*/@Scheduled(fixedDelay = 300)public void aaa() {Integer requestParam = (int) (Math.random() * 100) + 1;batchCollapser.addRequestParam(requestParam);System.out.println("当前请求参数:" + requestParam);}
}

@Data
public class Product {private Integer id;private String notes;
}

当然以上工具类仅仅只是DEMO，各位大佬可自行完善，权衡请求合并利弊，降低服务器在高并发请求时的压力

定时线程池实现请求合并相关推荐

java异步线程池同时请求多个接口数据
java异步线程池同时请求多个接口数据一.适合的使用场景复杂的网页爬虫,如要同时请求多个不同网页的数据,并且需要执行不同的数据处理,这个是非常合适的,执行线程传递的参数到最后callback是会附 ...
11.定时任务定时线程池详解
3.1 新增定时任务池 11.定时任务&定时线程池详解当我们不用任务框架时,我们想自己写一个定时任务时,我们能想起那个工具类呢?Timer ?还有吗?不知道了,下面我们要讲下Schedu ...
java定时线程池_java 定时器线程池（ScheduledThreadPoolExecutor）的实现
前言定时器线程池提供了定时执行任务的能力,即可以延迟执行,可以周期性执行.但定时器线程池也还是线程池,最底层实现还是ThreadPoolExecutor,可以参考我的另外一篇文章多线程–精通Thre ...
JAVA线程池_并发队列工作笔记0003---线程池的分类_可缓存线程池_定长线程池_定时线程池_单例线程池
技术交流QQ群[JAVA,C++,Python,.NET,BigData,AI]:170933152 这里说线程池的分类有可缓存类型, 定长类型, 定时类型, 单例类型, 这里我这次用Executo ...
并发编程定时线程池ScheduledThreadPoolExecutor学习总结
ScheduledThreadPoolExecutor 类结构图: 用来处理延时任务或定时任务,采用DelayQueue存储等待的任务,DelayQueue内部封装了一个PriorityQueue,它 ...
线程池处理高并发请求
背景本系统(支付系统)会在每个月特定时间(如账单日某个时间)接收上游系统发起的大量请求并进行处理,并在处理完成后返回结果给上游系统.而本系统接收到请求进行处理的过程是调用第三方(支付公司)进行处理并 ...
Android工程师进阶第五课多线程锁,线程池和DVM/ART优化
第09讲:Java 线程优化偏向锁,轻量级锁.重量级锁我目前所在的公司是一家跨国企业,总部在瑞典.前段时间公司新开发的一个应用准备发布到应用宝平台.但是在发布之前,需要准备一系列软著相关的证明材料 ...
Java线程池实现原理及其在美团业务中的实践
来自:美团技术团队随着计算机行业的飞速发展,摩尔定律逐渐失效,多核CPU成为主流.使用多线程并行计算逐渐成为开发人员提升服务器性能的基本武器.J.U.C提供的线程池ThreadPoolExecuto ...
线程池 And 线程池的使用(基于.net平台)
多线程可以提高应用程序的效率,这是肯定的,但是,效率是不是最优的呢,是不是觉得多线程很复杂呢? 前面学习线程的知道,用多线程需要CreateThread创建线程,还要关闭线程.另外,多线程有时候还要对 ...
由浅入深理解Java线程池及线程池的如何使用
前言多线程的异步执行方式,虽然能够最大限度发挥多核计算机的计算能力,但是如果不加控制,反而会对系统造成负担.线程本身也要占用内存空间,大量的线程会占用内存资源并且可能会导致Out of Memory ...

定时线程池实现请求合并

一、什么是请求合并

二、请求合并的优缺点

三、请求合并技术实现

定时线程池实现请求合并相关推荐

最新文章

热门文章