一劳永逸的优化！并发RPC调用小工具

眯眯眼的猫头鹰

发布于 2022-03-23 09:51:26

6530

发布于 2022-03-23 09:51:26

前言

系统的性能优化是每一个程序员的必经之路，但也可能是走过的最深的套路。它不仅需要对各种工具的深入了解，有时还需要结合具体的业务场景得出定制化的优化方案。当然，你也可以在代码中悄悄藏上一个Thread.sleep，在需要优化的时候少睡几毫秒（手动狗头）。性能优化这个课题实在是太浩瀚了，以至于目前市面上没有一本优质的书能够全面的总结这个课题。不仅如此，即使是深入到各个细分领域上，性能优化的手段也非常丰富，令人眼花缭乱。

本文也不会涵盖所有的优化套路，仅就最近项目开发过程中遇到的并发调用这一个场景给出自己的通用方案。大家可以直接打包或是复制粘贴到项目中使用。也欢迎大家给出更多的意见还有优化场景。

背景

不知大家在开发过程中是否遇到这样的一个场景，我们会先去调用服务A，然后调用服务B，组装一下数据之后再去调用一下服务C（如果你在微服务系统的开发中没有遇到这样的场景，我想说，要么你们系统的拆分粒度太粗，要么这一个幸运无下游服务依赖的底层系统~）

这条链路的耗时就是 duration(A) + duration(B) + duration(C) + 其它操作。从经验来看，大部分的耗时都来自于下游服务的处理耗时和网络IO，应用内部的CPU操作的耗时相比而言基本可以忽略不计。但是，当我们得知对服务A和B的调用之间是无依赖的时候，是否可以通过同时并发调用A和B来减少同步调用的等待耗时，这样理想情况下链路的耗时就可以优化成 max(duration(A),duration(B)) + duration(C) + 其它操作

再举一个例子，有时我们可能需要批量调用下游服务，比如批量查询用户的信息。下游查询接口出于服务保护往往会对单次可以查询的数量进行约束，比如一次只能查一百条用户的信息。因此我们需要多请求拆分多次进行查询，于是耗时变成了 n*duration(A) + 其它操作。同样，用并发请求的优化方式，理想情况下耗时可以降到 max(duration(A)) + 其它操作

这两种场景的代码实现基本类似，本文将会提供第二种场景的思路和完整实现。

小试牛刀

并发RPC调用的整体实现类图如下：

首先我们需要创建一个线程池用于并发执行。因为程序中通常还有别的使用线程池的场景，而我们希望RPC调用能够使用一个单独的线程池，因此这里用工厂方法进行了封装。

@Configuration
public class ThreadPoolExecutorFactory {

    @Resource
    private Map<String, AsyncTaskExecutor> executorMap;

    /**
    * 默认的线程池
    */
    @Bean(name = ThreadPoolName.DEFAULT_EXECUTOR)
    public AsyncTaskExecutor baseExecutorService() {
        //后续支持各个服务定制化这部分参数
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        //设置线程池参数信息
        taskExecutor.setCorePoolSize(10);
        taskExecutor.setMaxPoolSize(50);
        taskExecutor.setQueueCapacity(200);
        taskExecutor.setKeepAliveSeconds(60);
        taskExecutor.setThreadNamePrefix(ThreadPoolName.DEFAULT_EXECUTOR + "--");
        taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
        taskExecutor.setAwaitTerminationSeconds(60);
        taskExecutor.setDaemon(Boolean.TRUE);
        //修改拒绝策略为使用当前线程执行
        taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        //初始化线程池
        taskExecutor.initialize();

        return taskExecutor;
    }

    /**
    * 并发调用单独的线程池
    */
    @Bean(name = ThreadPoolName.RPC_EXECUTOR)
    public AsyncTaskExecutor rpcExecutorService() {
        //后续支持各个服务定制化这部分参数
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        //设置线程池参数信息
        taskExecutor.setCorePoolSize(20);
        taskExecutor.setMaxPoolSize(100);
        taskExecutor.setQueueCapacity(200);
        taskExecutor.setKeepAliveSeconds(60);
        taskExecutor.setThreadNamePrefix(ThreadPoolName.RPC_EXECUTOR + "--");
        taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
        taskExecutor.setAwaitTerminationSeconds(60);
        taskExecutor.setDaemon(Boolean.TRUE);
        //修改拒绝策略为使用当前线程执行
        taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        //初始化线程池
        taskExecutor.initialize();

        return taskExecutor;
    }
    /**
     * 根据线程池名称获取线程池
     * 若找不到对应线程池，则抛出异常
     * @param name 线程池名称
     * @return 线程池
     * @throws RuntimeException 若找不到该名称的线程池
     */
    public AsyncTaskExecutor fetchAsyncTaskExecutor(String name) {
        AsyncTaskExecutor executor = executorMap.get(name);
        if (executor == null) {
            throw new RuntimeException("no executor name " + name);
        }
        return executor;
    }
}

public class ThreadPoolName {

    /**
     * 默认线程池
     */
    public static final String DEFAULT_EXECUTOR = "defaultExecutor";

    /**
     * 并发调用使用的线程池
     */
    public static final String RPC_EXECUTOR = "rpcExecutor";
}

如代码所示，我们声明了两个Spring的线程池AsyncTaskExecutor，分别是默认的线程池和RPC调用的线程池，并将它们装载到map中。调用方可以使用fetchAsyncTaskExecutor方法并传入线程池的名称来指定线程池执行。这里还有一个细节，Rpc线程池的线程数要显著大于另一个线程池，是因为Rpc调用不是CPU密集型逻辑，往往伴随着大量的等待。因此增加线程数量可以有效提高并发效率。

@Component
public class TracedExecutorService {

    @Resource
    private ThreadPoolExecutorFactory threadPoolExecutorFactory;


    /**
     * 指定线程池提交异步任务，并获得任务上下文
     * @param executorName 线程池名称
     * @param tracedCallable 异步任务
     * @param <T> 返回类型
     * @return 线程上下文
     */
    public <T> Future<T> submit(String executorName, Callable<T> tracedCallable) {
        return threadPoolExecutorFactory.fetchAsyncTaskExecutor(executorName).submit(tracedCallable);
    }
}

submit方法封装了获取线程池和提交异步任务的逻辑。这里采用Callable+Future的组合来获取异步线程的执行结果。

线程池准备就绪，接着我们就需要声明一个接口用于提交并发调用服务：

public interface BatchOperateService {

    /**
     * 并发批量操作
     * @param function 执行的逻辑
     * @param requests 请求
     * @param config 配置
     * @return 全部响应
     */
    <T, R> List<R> batchOperate(Function<T, R> function, List<T> requests, BatchOperateConfig config);
}

@Data
public class BatchOperateConfig {

    /**
     * 超时时间
     */
    private Long timeout;

    /**
     * 超时时间单位
     */
    private TimeUnit timeoutUnit;

    /**
     * 是否需要全部执行成功
     */
    private Boolean needAllSuccess;

}

batchOperate方法中传入了function对象，这是需要并发执行的代码逻辑。requests则是所有的请求，并发调用会递归这些请求并提交到异步线程。config对象则可以对这次并发调用做一些配置，比如并发查询的超时时间，以及如果部分调用异常时整个批量查询是否继续执行。

接下来看一看实现类：

@Service
@Slf4j
public class BatchOperateServiceImpl implements BatchOperateService{

    @Resource
    private TracedExecutorService tracedExecutorService;

    @Override
    public <T, R> List<R> batchOperate(Function<T, R> function, List<T> requests, BatchOperateConfig config) {
        log.info("batchOperate start function:{} request:{} config:{}", function, JSON.toJSONString(requests), JSON.toJSONString(config));

        // 当前时间
        long startTime = System.currentTimeMillis();

        // 初始化
        int numberOfRequests = CollectionUtils.size(requests);

        // 所有异步线程执行结果
        List<Future<R>> futures = Lists.newArrayListWithExpectedSize(numberOfRequests);
        // 使用countDownLatch进行并发调用管理
        CountDownLatch countDownLatch = new CountDownLatch(numberOfRequests);
        List<BatchOperateCallable<T, R>> callables = Lists.newArrayListWithExpectedSize(numberOfRequests);

        // 分别提交异步线程执行
        for (T request : requests) {
            BatchOperateCallable<T, R> batchOperateCallable = new BatchOperateCallable<>(countDownLatch, function, request);
            callables.add(batchOperateCallable);

            // 提交异步线程执行
            Future<R> future = tracedExecutorService.submit(ThreadPoolName.RPC_EXECUTOR, batchOperateCallable);
            futures.add(future);
        }

        try {
            // 等待全部执行完成，如果超时且要求全部调用成功，则抛出异常
            boolean allFinish = countDownLatch.await(config.getTimeout(), config.getTimeoutUnit());
            if (!allFinish && config.getNeedAllSuccess()) {
                throw new RuntimeException("batchOperate timeout and need all success");
            }
            // 遍历执行结果，如果有的执行失败且要求全部调用成功，则抛出异常
            boolean allSuccess = callables.stream().map(BatchOperateCallable::isSuccess).allMatch(BooleanUtils::isTrue);
            if (!allSuccess && config.getNeedAllSuccess()) {
                throw new RuntimeException("some batchOperate have failed and need all success");
            }

            // 获取所有异步调用结果并返回
            List<R> result = Lists.newArrayList();
            for (Future<R> future : futures) {
                R r = future.get();
                if (Objects.nonNull(r)) {
                    result.add(r);
                }
            }
            return result;
        } catch (Exception e) {
            throw new RuntimeException(e.getMessage());
        } finally {
            double duration = (System.currentTimeMillis() - startTime) / 1000.0;
            log.info("batchOperate finish duration:{}s function:{} request:{} config:{}", duration, function, JSON.toJSONString(requests), JSON.toJSONString(config));

        }
    }
}

通常我们提交给线程池后直接遍历Future并等待获取结果就好了。但是这里我们用CountDownLatch来做更加统一的超时管理。可以看一下BatchOperateCallable的实现：

public class BatchOperateCallable<T, R> implements Callable<R> {

    private final CountDownLatch countDownLatch;

    private final Function<T, R> function;

    private final T request;

    /**
     * 该线程处理是否成功
     */
    private boolean success;

    public BatchOperateCallable(CountDownLatch countDownLatch, Function<T, R> function, T request) {
        this.countDownLatch = countDownLatch;
        this.function = function;
        this.request = request;
    }

    @Override
    public R call() {
        try {
            success = false;
            R result = function.apply(request);
            success = true;
            return result;
        } finally {
            countDownLatch.countDown();
        }
    }

    public boolean isSuccess() {
        return success;
    }
}

无论调用时成功还是异常，我们都会在结束后将计数器减一。当计数器被减到0时，则代表所有并发调用执行完成。否则如果在规定时间内计数器没有归零，则代表并发调用超时，此时会抛出异常。

潜在问题

并发调用的一个问题在于我们放大了访问下游接口的流量，极端情况下甚至放大了成百上千倍。如果下游服务并没有做限流等防御性措施，我们极有可能将下游服务打挂（这种原因导致的故障屡见不鲜）。因此需要对整个并发调用做流量控制。流量控制的方法有两种，一种是如果微服务采用mesh的模式，则可以在sidecar中配置RPC调用的QPS，从而做到全局的管控对下游服务的访问（这里选择单机限流还是集群限流取决于sidecar是否支持的模式以及服务的流量大小。通常来说平均流量较小则建议选择单机限流，因为集群限流的波动性往往比单机限流要高，流量过小会造成误判）。如果没有开启mesh，则需要在代码中自己实现限流器，这里推荐Guava的RateLimiter类，但是它只支持单机限流，如果要想实现集群限流，则方案的复杂度还会进一步提升

小结

将项目开发中遇到的场景进行抽象并尽可能的给出通用的解决方案是我们每一个开发者自我的重要方式，也是提高代码复用性和稳定性的利器。并发Rpc调用是一个常见解决思路，希望本文的实现可以对你有帮助。

本文参与腾讯云自媒体分享计划，分享自作者个人站点/博客。

原始发表：2021-10-04，如有侵权请联系 cloudcommunity@tencent.com 删除

微服务

rpc

本文分享自作者个人站点/博客前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体分享计划，欢迎热爱写作的你一起参与！

微服务

rpc

登录后参与评论

0 条评论

热度