接口请求合并技巧,用好了效率直接翻倍!
Java架构师社区
共 12584字,需浏览 26分钟
·
2023-10-06 21:35
一、什么是请求合并
在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
* @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
* @desc
**/
@Service
public class ProductService implements BatchCollapser.BatchHandler<List<Integer>, Integer> {
private BatchCollapser<Integer, Integer> batchCollapser;
@PostConstruct
private void postConstructorInit() {
// 当请求数量达到20个,或每过5s合并执行一次请求
batchCollapser = BatchCollapser.getInstance(ProductService.this, 20, 5);
}
@Override
public 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,各位大佬可自行完善,权衡请求合并利弊,降低服务器在高并发请求时的压力
来源:blog.csdn.net/leilei1366615/ article/details/123858619
这些年小编给你分享过的干货
转发在看就是最大的支持❤️
评论