从单机到分布式微服务，大文件校验上传的通用解决方案

一、先说结论

本文将结合我的工作实战经历，总结和提炼一种从单体架构到分布式微服务都适用的一种文件上传和校验的通用解决方案，形成一个完整的方法论。本文主要解决手段包括多线程、设计模式、分而治之、MapReduce等，虽然文中使用的编程语言为Java，但解决问题和优化思路是互通的，适合有一定开发经验的开发者阅读，希望对大家有帮助。

二、引言

文件上传的场景应该都不陌生，不管是C端还是B端，都会有文件上传的场景。用户在平台页面点击上传文件，用户请求在最后会到达后端服务器，后端服务器会对上传的文件进行各种校验，比如文件名称校验、文件大小校验、文件内容校验等，其中业务逻辑最复杂、技术上有挑战性的当属文件内容校验了。为什么这么说呢？接着看。

三、背景

文件校验和上传，看似是一件很简单的工作，要做好，可能也并非一件容易得事情。我以一个电商后台系统为例，上传csv格式的sku信息文档将会面临下面几方面挑战：

上传sku数量多：上传文件中sku数量不定，从个位数到百万级不等；为了好的用户体验，需要在较短的时间内上传校验完成并返回结果；
业务逻辑复杂：文件上传校验需要校验每条内容，校验规则多且复杂，校验规则包括录入的sku格式是否符合，如不符合需要给出提示语1；校验上传的sku是否合法有效，如果需要给出相应的提示语2；校验该操作人是否有该sku管理权限，如果没有给出相应的提示语3……每个校验逻辑中可能还包含许多分支、循环逻辑……
外部依赖RPC多：上传校验过程中涉及多个外部依赖RPC的调用，比如sku的管理权限校验，需要调用用户中台RPC接口获取上传人的基本信息；校验sku是否是本次活动范围，需要调用直播中台RPC接口……

四、关键问题拆解和解决思路

上传数量多且要求体验友好，就要求要注意高性能方面的优化：对于业务服务器来说，如果是单机性能优化，需要考虑使用多线程技术来充分发挥服务器性能；如果是分布式的服务，在优化单机性能无法业务场景需要的时候，还可以考虑依靠中间件来协同不同服务器，发挥集群优势。
业务逻辑复杂，就要求写出来的代码有较高的可阅读性、可维护性，不要成为“大泥球”：除了在系统架构方面的优化之外，对于开发人员，可以考虑使用设计模式来提高代码质量。
外部RPC依赖多，网络数据IO操作，接口性能可能无法保证，就需要使用异步调用的方式来保证性能；

五、系统架构

假设有这么一个电商活动管理系统，从架构上来说，可以分为服务层、业务层、数据层和外部依赖，架构图如下：

服务层：包括对外服务和外部调用；
业务层：活动的生命周期，包括创建、查看、修改、关闭流程；
数据层：数据存储，主要是数据库集群和缓存集群；
外部依赖：外部依赖的RPC服务，包括商品RPC服务等；

在技术实现方面，该系统是前后端分离的系统，前后端通过域名进行交互。前端服务主要提供操作页面，用户可以在页面端进行各种操作，例如创建活动、查看活动、修改活动、关闭活动等；

后端采用的是微服务架构，按照功能拆分为提供HTTP接口的soa应用、提供MQ消费功能的MQ应用、提供RPC服务的RPC应用，存储使用的是MySQL和Redis集群，大概架构图如下：

六、Java多线程实践

6.1 使用Java多线程优化单机性能

分析上面的场景，明显是IO密集型的场景。IO 密集型指的是大部分时间都在执行 IO 操作，主要包括网络 IO 和磁盘 IO，以及与计算机连接的一些外围设备的访问。在上面场景中，校验过程中需要调用大量RPC接口，大部分时间调用都在等待网络IO，所以可以使用异步和多线程的设计方法来提升网络IO性能，从而优化整体性能。

关于Java多线程在这里不赘述了，直接看关键代码实现吧：

    ExecutorService executorService = Executors.newFixedThreadPool(10);@ResponseBody@RequestMapping(value = "uploadSku", method = RequestMethod.POST)public Result uploadSku(@RequestParam(value = "file", required = false) MultipartFile file) throws IOException {Result result = new Result();result.setSuccess(true);BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(file.getInputStream()));try {// 校验文件名称result = checkFileNameFormat(file);if (!result.isSuccess()) {return result;}// 校验文件内容格式并填充校验任务List<UploadResInfo> uploadResInfos = new ArrayList<>();List<SkuCheckTask> tasks = checkFileContentAndFillSkuCheckTask(result, bufferedReader, uploadResInfos);// 执行校验任务result = dealSkuSkuCheckTask(tasks, uploadResInfos);} catch (Exception e) {result.setSuccess(false);result.setErrorMessage("上传文件异常！");}return result;}/*** @param tasks* @param uploadResInfos* @return*/private Result dealSkuSkuCheckTask(List<SkuCheckTask> tasks, List<UploadResInfo> uploadResInfos) throws Exception {Result result = new Result();result.setSuccess(true);List<Long> passedSkus = new ArrayList<>();if (!CollectionUtils.isEmpty(tasks)) {List<Future<Result>> futureList = executorService.invokeAll(tasks);for (Future<Result> tempResult : futureList) {if (tempResult.get().isSuccess()) {Result tempRes = tempResult.get();if (null != tempRes.getResult().get("uploadResInfos")) {uploadResInfos.addAll((List<UploadResInfo>) tempRes.getResult().get("uploadResInfos"));}passedSkus.addAll((List<Long>) tempRes.getObject());}}}result.addDefaultModel("passedSkus", passedSkus);if (passedSkus.size() == 0) {result.setErrorMessage("上传都不通过");}return result;}

public class SkuCheckTask implements Callable<Result> {private List<Long> skuList;public SkuCheckTask(List<Long> skuList) {this.skuList = skuList;}@Overridepublic Result call() throws Exception {Result result = new Result();result.setSuccess(true);List<Long> passedSkuList = new ArrayList<>();List<UploadResInfo> uploadResInfos = new ArrayList<>();for (int i = 0; i < skuList.size(); i++) {if (checkSku(skuList.get(i))) {passedSkuList.add(skuList.get(i));} else {UploadResInfo uploadResInfo = new UploadResInfo(skuList.get(i).toString(), false, "RPC校验失败");uploadResInfos.add(uploadResInfo);}}result.setObject(passedSkuList);result.addDefaultModel("uploadResInfos", uploadResInfos);return result;}/*** 校验sku，复杂校验逻辑** @param sku* @return*/private boolean checkSku(Long sku) {// 复杂校验逻辑，例如多个RPC调用等耗时操作System.out.println("校验sku：" + sku);return true;}
}

6.2 线程数的设置

我们知道，调整线程池中的线程数量的主要是为了充分并合理地使用 CPU 和内存等资源，从而最大限度地提高程序的性能。

对于CPU密集型任务（比如加解密、压缩和解压、计算），最佳的线程数为 CPU 核心数的 1~2 倍，如果设置过多的线程数，实际上并不会起到很好的效果。因为CPU密集型任务本来就会占用大量的CPU资源，CPU 的每个核心工作基本都是满负荷的，而如果设置了过多的线程，每个线程都要去争取CPU资源来执行自己的任务，这就会造成不必要的上下文切换，此时线程数的增多反而会导致性能下降。

对于IO密集型任务（比如数据库读写、文件读写、网络通信等），这种任务并不会太消耗CPU资源，反而是在等待IO操作。线程数设置可以参考以下公式：

线程数 = CPU核心数 * （1 + 平均等待时间/平均工作时间）

在本程序中，使用了线程池：FixedThreadPool，并将线程数设置为10。这里的考虑是容器为16C32G的配置，除了上传任务，服务端还会处理其他的任务，还有其他的线程池，为了综合考虑，这里只是分配了10个线程数。当然，最佳实践是使用远程配置中心动态调整线程池线程数，实现动态线程池，在实践中进行调整和压测，最终找到合适的线程数配置。

七、责任链模式实践

对于上述这个校验逻辑，最常见的处理方式是使用 if…else…条件判断语句来处理，这样处理可能存在这样的问题：

代码复杂度高：该场景中的判定条件通常不是简单的判断，需要调用外部RPC接口查询数据，从结果中解析到需要的字段，才能进行逻辑判断。这样代码的嵌套层数就会很多，代码复杂度就会很高，不用太久，这段代码将发展成为“大泥球”。
代码耦合度高：如果业务需求新增校验逻辑，那么就要继续添加 if…else…判定条件；另外，这个条件判定的顺序也是写死的，如果想改变顺序，那么也只能修改这个条件语句。

那么面对上面这种场景，如何实现更优雅呢？。其实这里也很简单，就是把判定条件的部分放到处理类中，这就是责任链模式。如果满足条件 1，则由 Handler1 来处理，不满足则向下传递；如果满足条件 2，则由 Handler2 来处理，不满足则继续向下传递，以此类推，直到条件结束。部分代码如下：

Handler接口：

public interface SkuCheckHandler {BaseResult doHandler(UploadInfo uploadInfo);
}

SkuCheckHandler接口实现Handler1：

public class Handler1 implements SkuCheckHandler {@Overridepublic BaseResult doHandler(UploadInfo uploadInfo) {// 调用用户中台校验权限return new BaseResult();}
}

遍历Handler进行校验，如果Handler校验不通过直接返回校验结果，校验通过则继续进入下一个Handler进行校验：

public class SkuCheckHandlerChain {private List<SkuCheckHandler> handlers = new ArrayList<>();public void addHandler(SkuCheckHandler skuCheckHandler) {this.handlers.add(skuCheckHandler);}public BaseResult handle(UploadInfo uploadInfo){BaseResult baseResult = new BaseResult();baseResult.setSuccess(true);for (SkuCheckHandler handler : handlers) {baseResult = handler.doHandler(uploadInfo);if (!baseResult.isSuccess()) {return baseResult;}}return baseResult;}}

责任链设置和调用：

    private boolean checkSku(Long sku) {// 复杂校验逻辑，例如多个RPC调用等耗时操作System.out.println("校验sku：" + sku);// 后续校验都依赖商品信息，所以需要调商品RPC获取Sku信息-uploadInfoUploadInfo uploadInfo = new UploadInfo();SkuCheckHandlerChain handlerChain = new SkuCheckHandlerChain();handlerChain.addHandler(new Handler1());handlerChain.addHandler(new Handler2());BaseResult baseResult = handlerChain.handle(uploadInfo);return baseResult.isSuccess();}

八、分布式文件上传最佳实践

8.1 MapReduce简介

当使用了多线程技术，并优化了线程数，似乎单机性能已经达到了极限。但是如果此时仍然不能满足业务场景需要，那又该怎么优化呢？

有人可能会想到垂直扩容，升级更高配的机器来提升性能。这个办法当然是可行的，也是最简单粗暴的方式，唯一的缺点就是“费钱”，土豪请随意。一般来说，Google的方式可能更加值得借鉴，Google使用“3M胶带粘在一起的服务器”打败了成本更高的高配计算机。

在面对海量数据背景下，Google科学家杰夫·迪恩提出了MapReduce技术。MapReduce其实并不复杂，使用的正是分而治之（Divide and Conquer）的思想。打个不太恰当的比方就是，老板分作业，小兵完成作业，老板进行汇总。

MapReduce其实也是自顶向下的递归。MapReduce先在最顶层将一个复杂的大任务分解成为成百上千个小任务；然后将每个小任务分配到一个服务器上去求解；最后再将每个服务器上面的结果综合起来，得到原来大任务的最终结果。第一个自顶向下分解的过程称为Map，第二个自底向上合并的过程称为Reduce。

其核心原理其实可以看这张图，图片出自论文《MapReduce: Simplified Data Processing on Large Clusters》。

8.2 MapReduce在文件上传场景的应用

单机服务器性能无法满足，应该考虑合理利用多台机器，不同微服务之间相互协作，共同完成上传的任务。借鉴MapReduce核心思想，可以使用现有系统架构，实现大文件的分布式上传和校验。

一图胜前言，方案说明都在图片中了，详细请看：

九、踩坑和代码调试

9.1 踩坑1：MQ消费中使用LoginContext获取用户信息异常

其中有个踩坑点需要注意，在soa应用中常用的LoginContext获取用户信息；在MQ应用中，使用LoginContext将无法获取到用户信息，如果使用将会出现空指针异常；出现异常之后，MQ消费将会进行重试，重试也一直会发生异常，从而死循环，无法得到正确的结果。

9.2 代码调试-Idea远程Debug

在开发工作中，代码写完并不是万事大吉了。部署到服务器测试过程中，可能还会发现各种各样意料之外的错误。当服务器日志打印过多或者过少都影响问题排查的效率，以文件上传场景为例，如果不打印完整的出入参，出现问题没有日志可以用来排查问题；如果每个方法都打印完整的出入参日志，当上传文件中sku数量较多，可以想象下如果有100w条的sku信息，从这么多的日志中去排查问题无异于“大海捞针”。

那这个问题无解了吗？当然不是，远程Debug可以提升排查效率，同事妹子看见了都直呼YYDS。其实这个工具就是我们几乎人人都在用的Idea，Idea自带了远程调试工具。下面是我的使用经验，适用于部署在Tomcat容器工程代码：

9.2.1 环境配置

远程Tomcat配置

远程Tomcat添加启动参数并重启生效：

-agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=5005

Idea配置

话不多说，图上都有:

启动调试

9.2.2 常见问题

为什么调试断点没生效？

本地和远程代码要相同，不一样则会出现无法进入断点的情况；如果代码一致还是无法进入，尝试重启，一般可以解决；

进入断点调试之后，服务器还可以处理其他请求吗？

服务器在断点处停住了，无法处理其他请求；

改了本地代码可以直接debug吗？

不可以，需要部署在远程服务器之后再次启动debug；

通用解决方案总结

通过上述过程之后，总结出一套通用的大文件上传和校验的解决方案。总结一下就是，如果现在技术架构还处在单机架构的阶段，可以考虑使用多线程技术优化单机性能；为了使代码优雅一点，可以考虑使用责任链模式；如果现在技术架构已经发展到分布式和微服务了，可以借鉴分而治之的思想，让多服务器协作工作，发挥多服务器的优势。

如果用三个词总结，那就是：多线程、责任链模式、分而治之和MapReduce。