collapse executor 是一个高性能、低延迟的批量执行器,可有效支持高并发的热点请求,支持与Spring Boot集成,帮助开发者快速构建高性能的微服务,提高服务资源利用率的同时降低服务响应时间。涉及技术点包括CompletableFuture、Spring Boot、WebClient、Servlet Async等主流技术栈
- 部分接口的并发量很高,想提升性能的同时又不希望引入缓存服务,减少额外维护成本。
- 单次调用转批量调用。在同一时刻将不同线程的参数合并为一个参数再执行批处理逻辑。
- 希望降低客户端的远程连接数量。
- 希望减少服务端工作线程消耗。
- 批量获取基于DB/redis的自增序列
- API简单易上手,扩展难度低
默认提供了
spring-boot集成,可在spring-boot环境下开箱即用。且核心逻辑已高度抽象,二次扩展实现简单,仅需编写合并请求以及拆分响应两块逻辑。
- 高性能0延迟,发起批量请求时无需等待时间窗口
巧妙的利用了
提交任务->任务执行两个行为之间的时间间隔进行输入的批量收集,相比于等待一个时间窗口(如等待2ms)的设计,该设计可在批处理下依然保证极高的实时性,且在高并发的场景下也有着不俗的批量收集能力。
- 设计简单易维护,无需维护第三方服务
核心逻辑完全不依赖任何第三方库/服务,完全基于JDK库进行实现。
以下两张图解释了有无折叠执行器的调用差异。当无请求折叠时,请求与网络连接数的比例为1:1;当使用请求折叠后,请求与网络连接数的比例为N:1,即多个请求会合并为一个请求发起远程调用,由此可以做到减少I/O次数、减少后端压力,从而提升调用性能降低RT。
必备条件: JDK8及以上
该方式适用于简单的幂等请求的场景,通常需要用户手动指定本次调用所属的并发分组。
以下该案例表示将当前传入的Callable按照
example group进行分组。 同一并发分组下的Callable仅执行一次,并将这一次的返回结果作为同一并发分组发起的请求结果
BlockingCollapseExecutorExample.java
public class BlockingCollapseExecutorExample { public static void main(String[] args) throws Throwable { BlockingCallableGroupCollapseExecutor blockingCollapseExecutor = new BlockingCallableGroupCollapseExecutor(); String outputString = blockingCollapseExecutor.execute("example group", () -> "Hello World Collapse Executor. Blocking"); System.out.println(outputString); } }
AsyncCollapseExecutorExample.java
public class AsyncCollapseExecutorExample { public static void main(String[] args) throws Throwable { AsyncCallableGroupCollapseExecutor asyncCallableGroupCollapseExecutor = new AsyncCallableGroupCollapseExecutor(); asyncCallableGroupCollapseExecutor.setExecutor(new ThreadPoolExecutor(10, 10, 0, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), r -> { Thread thread = new Thread(r); thread.setDaemon(true); return thread; })); asyncCallableGroupCollapseExecutor.execute("example group", () -> "Hello World Collapse Executor. Async") .thenAccept(System.out::println) .thenRun(() -> System.exit(0)); System.in.read(); } }
FutureCollapseExecutorExample.java
这种方式必须保证Callable中的处理逻辑是非阻塞的!!!
public class FutureCollapseExecutorExample { public static void main(String[] args) throws Throwable { FutureCallableGroupCollapseExecutor futureCollapseExecutor = new FutureCallableGroupCollapseExecutor(); futureCollapseExecutor.execute("example group", () -> CompletableFuture.completedFuture("Hello World Collapse Executor. Future")) .thenAccept(System.out::println) .thenRun(() -> System.exit(0)); System.in.read(); } }
该方式适用于后端服务提供了批处理接口的场景,将同并发下其他线程的输入合并调用后端服务的批处理接口,可以减少多次不必要的单次调用,如批量查询。
由于这种方式可以更好的处理输入组,故该方式合并效率可以更高,由此带来的性能提升也会更高。
CustomBlockingCollapseExecutor.java
public class CustomBlockingCollapseExecutor extends CollapseExecutorBlockingSupport<Long, UserEntity, Map<Long, UserEntity>> { @Override protected Map<Long, UserEntity> doExecute(Collection<Input<Long>> inputs) { //此处编写批量请求逻辑 return null; } @Override protected void bindingOutput(Map<Long, UserEntity> users, List<Bundle<Long, UserEntity>> bundles) { //此处编写批量响应与原始请求关联的逻辑 } }
CustomAsyncCollapseExecutor.java
与同步阻塞调用类似,主要差异为需要设置一个异步线程池,用于执行批量请求逻辑。
public class CustomBlockingCollapseExecutor extends CollapseExecutorAsyncSupport<Long, UserEntity, Map<Long, UserEntity>> { @Override protected Map<Long, UserEntity> doExecute(Collection<Input<Long>> inputs) { //此处编写批量请求逻辑 return null; } @Override protected void bindingOutput(Map<Long, UserEntity> users, List<Bundle<Long, CompletableFuture<UserEntity>>> bundles) { for (Bundle<Long, CompletableFuture<UserEntity>> bundle : bundles) { Long inputId = bundle.getInput(); UserEntity userEntity = users.get(inputId); //需要返回CompletableFuture类型 bundle.bindOutput(CompletableFuture.completedFuture(userEntity)); } } }
collapse executor已经对Spring Boot进行了适配,利用Spring Boot自动装配能力简化了在Spring环境下的使用体验,通过简单的配置即可使用。
在Spring Boot中,collapse executor对主要的几个组件进行了适配,分别是
RestTemplate
、WebClient
、Servlet。
以下是一个折叠执行器的yaml配置例子及解释,详情可参考collapse-executor-sample-spring-boot中的application.yaml
collapse: executor: enabled: true # 折叠执行器的总开关,配置为false后,后面的所有配置(servlet、rest-template、web-client)将失效 wait-threshold: 10 #批量收集的最小阈值 collecting-wait-time: 0 #声明批量收集未达到阈值时的行为。 #collecting-wait-time < 0时:不做任何等待,立即发起调用 #collecting-wait-time = 0时:让出当前收集线程时间片等待下次调度后再发起调用 #collecting-wait-time > 0时:等待指定的时间后再发起调用,单位为毫秒(ms) rest-template: enabled: true #true表示打开RestTemplate的合并拦截器 collapse-policies: #声明合并策略,可以配置多个 sample-policy1: #策略名字 collapse-request-headers: #声明需要合并的请求头名字 - authorization collapse-request-queries: #声明需要合并的查询参数名字 - sample sample-policy2: collapse-request-headers: - user-id collapse-request-queries: - sample collapse-groups: # collapse-policy-name可以省略,省略后使用默认策略仅合并path相同的请求,而忽略其他任何参数 # 例如:此时并发发起 /user/2、/user/2、/article/2、/article/2 [4]个请求,由于前两个请求满足 /user/*,则会将前两个合并为 [1] 个请求发起调用; # 而第三第四个/article/2请求没有匹配到配置中的声明的折叠组,则依然会按照 [2] 个请求分别发起调用 - uris: - /user/* - /test/noop* #------------------------------------------------------------------ # 例如:此时并发发起 /samples/1(header:authorization=test), /samples/1(header:authorization=test), /samples/1(header:authorization=demo) [3]个请求, # 由于前两个请求携带的[authorization]请求头值相同,则会将前两个合并为 [1] 个请求发起调用; # 而第三个请求则会单独发起调用,与前两个不是同一组! - collapse-policy-name: sample-policy1 #需要与前面声明的策略名对应 uris: - /samples/*
@Configuration public class SampleConfiguration { /** * 基于RestTemplate的使用方式 */ @Bean public RestTemplate restTemplate(CollapseHttpRequestInterceptor collapseHttpRequestInterceptor) { //注入CollapseHttpRequestInterceptor拦截器实例,并添加到RestTemplate实例上 return new RestTemplateBuilder() .interceptors(collapseHttpRequestInterceptor) .build(); } /** * 基于WebClient的使用方式 */ @Bean public WebClient webClient(CollapseExchangeFilterFunction exchangeFilterFunction) { //注入CollapseExchangeFilterFunction过滤器实例,并添加到WebClient实例上 return WebClient.builder().filter(exchangeFilterFunction).build(); } }
3. 启动SpringBootSampleApplication查看结果
业务逻辑位于 AbstractBlockingCallSample
public abstract class AbstractBlockingCallSample { //已省略前后无关代码 @EventListener(ApplicationReadyEvent.class) public void processOnStarted() { UriComponentsBuilder baseUriBuilder = UriComponentsBuilder.fromUriString("http://localhost").port(serverPort); try { System.out.println("--------------------------------[" + prefix + "] start----------------------------------"); //查询id为1-50之间的所有用户,预期打印结果为执行了50次 queryId1between50(executorService, baseUriBuilder); //批处理查询id为1-50之间的所有用户,预期打印结果小于50次 queryId1between50Batch(); //单条查询id为1-2之间的所有用户,预期打印结果小于50次 queryId1between2(executorService, baseUriBuilder); System.out.println("--------------------------------[" + prefix + "] end------------------------------------\n\n\n"); } catch (InterruptedException e) { throw new RuntimeException(e); } } }
结论:相比于未使用折叠,TPS提升高达
94%
服务参数
server.tomcat.threads.max=200
服务地址(位于 collapse-executor-samples 中的 StressTestController, 后端均延迟[100ms]后响应,逻辑一致无差异)
http://localhost:8080/test/collapse100
http://localhost:8080/test/noop100
测试参数
400用户线程数,持续压测5分钟
http://localhost:8080/test/collapse100 启用请求折叠测试结果
TPS 3785/s
RT99 115ms
http://localhost:8080/test/noop100 关闭请求折叠测试结果
TPS 1951/s
RT99 211ms
结论:相比于未使用折叠,TPS提升高达
80%
RestTemplate默认配置,差异仅为是否包含折叠执行拦截器(CollapseHttpRequestInterceptor)
RestTemplate调用地址http://localhost:8080/test/noop0
200用户线程数,持续压测1分钟
启用请求折叠测试结果
TPS 15282/s
RT99 59ms
启用请求折叠测试结果
TPS 8455/s
RT99 180ms
结论:相比于未使用折叠,TPS提升高达
43%
WebClient默认配置,差异仅为是否包含折叠执行拦截器(CollapseExchangeFilterFunction)
WebClient调用地址http://localhost:8080/test/noop0
200用户线程数,持续压测1分钟
启用请求折叠测试结果
TPS 14276/s
RT99 35ms
启用请求折叠测试结果
TPS 9971/s
RT99 111ms
