Spring异步处理MultipartFile时临时文件被删除报错排查

问题现象 #

现有Spring应用包含多文件上传接口，为提升处理效率，创建了ThreadPool线程池，通过异步方式在独立线程中处理每个上传文件。参考同类实现教程编写测试代码后，运行未达到预期效果。
实际测试结果：

• 提交到线程池的任务数小于等于可用线程数时，程序运行完全正常
• 任务数大于可用线程数时，抛出如下异常：

java.io.FileNotFoundException: /private/var/folders/41/81526n295q1cptcb1tbrs544h504m8/T/tomcat.9191.5294201821824312569/work/Tomcat/localhost/ROOT/upload_ff67c7fe_2f2f_44c6_8eb9_3250c8a8739b_00000003.tmp (No such file or directory)
    at java.base/java.io.FileInputStream.open0(Native Method) ~[na:na]
    at java.base/java.io.FileInputStream.open(FileInputStream.java:216) ~[na:na]
    at java.base/java.io.FileInputStream.<init>(FileInputStream.java:157) ~[na:na]
    at org.apache.tomcat.util.http.fileupload.disk.DiskFileItem.getInputStream(DiskFileItem.java:198) ~[tomcat-embed-core-9.0.63.jar:9.0.63]
    at org.apache.catalina.core.ApplicationPart.getInputStream(ApplicationPart.java:100) ~[tomcat-embed-core-9.0.63.jar:9.0.63]
    at org.springframework.web.multipart.support.StandardMultipartHttpServletRequest$StandardMultipartFile.getInputStream(StandardMultipartHttpServletRequest.java:251) ~[spring-web-5.3.20.jar:5.3.20]
    at com.mhndev.springexecutor.service.UserService.parseCSVFile(UserService.java:47) ~[classes/:na]
    at com.mhndev.springexecutor.service.UserService.saveUsers(UserService.java:29) ~[classes/:na]
    at com.mhndev.springexecutor.service.UserService$$FastClassBySpringCGLIB$$c14fedc2.invoke(<generated>) ~[classes/:na]
    at org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:218) ~[spring-core-5.3.20.jar:5.3.20]
    at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:793) ~[spring-aop-5.3.20.jar:5.3.20]
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163) ~[spring-aop-5.3.20.jar:5.3.20]
    at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:763) ~[spring-aop-5.3.20.jar:5.3.20]
    at org.springframework.aop.interceptor.AsyncExecutionInterceptor.lambda$invoke$0(AsyncExecutionInterceptor.java:115) ~[spring-aop-5.3.20.jar:5.3.20]
    at org.springframework.aop.interceptor.AsyncExecutionAspectSupport.lambda$doSubmit$3(AsyncExecutionAspectSupport.java:278) ~[spring-aop-5.3.20.jar:5.3.20]
    at java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1768) ~[na:na]
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) ~[na:na]
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) ~[na:na]
    at java.base/java.lang.Thread.run(Thread.java:833) ~[na:na]

初步判断问题原因：当任务在线程池队列中等待分配线程时，MultipartFile对应的临时文件已经被提前删除。
推测参考教程未出现该错误的原因是，教程测试场景中上传文件数量始终与可用线程数匹配，没有出现任务排队等待的情况。

现有代码实现 #

Controller层代码 #

package com.mhndev.springexecutor.controller;

import com.mhndev.springexecutor.entity.User;
import com.mhndev.springexecutor.service.UserService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.multipart.MultipartFile;

import java.util.List;
import java.util.concurrent.CompletableFuture;

@RestController
public class UserController {

    @Autowired
    private UserService userService;

    @PostMapping(value = "/users", consumes = {MediaType.MULTIPART_FORM_DATA_VALUE})
    public ResponseEntity saveUsers(@RequestParam(value = "files") MultipartFile[] files) throws Exception {
        for(MultipartFile file: files) {
            userService.saveUsers(file);
        }

        return ResponseEntity.status(HttpStatus.CREATED).build();
    }

    @GetMapping(value = "/users", produces = "application/json")
    public CompletableFuture<ResponseEntity> findAllUsers() {
        return userService.findAllUsers().thenApply(ResponseEntity::ok);
    }

    @GetMapping(value = "/getUsersByThread", produces = "application/json")
    public  ResponseEntity getUsers(){
        CompletableFuture<List<User>> users1=userService.findAllUsers();
        CompletableFuture<List<User>> users2=userService.findAllUsers();
        CompletableFuture<List<User>> users3=userService.findAllUsers();
        CompletableFuture.allOf(users1,users2,users3).join();
        return ResponseEntity.status(HttpStatus.OK).build();
    }
}

Service层代码 #

package com.mhndev.springexecutor.service;

import com.mhndev.springexecutor.entity.User;
import com.mhndev.springexecutor.repository.UserRepository;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.web.multipart.MultipartFile;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;

@Service
public class UserService {

    @Autowired
    private UserRepository repository;
    Logger logger = LoggerFactory.getLogger(UserService.class);

    @Async("taskExecutor")
    public CompletableFuture<List<User>> saveUsers(MultipartFile file) throws Exception {
        long start = System.currentTimeMillis();
        List<User> users = parseCSVFile(file);
        logger.info("saving list of users of size {}, thread name : {}", users.size(), "" + Thread.currentThread().getName());
        users = repository.saveAll(users);
        long end = System.currentTimeMillis();
        logger.info("Total time {}", (end - start));
        return CompletableFuture.completedFuture(users);
    }

    @Async("taskExecutor")
    public CompletableFuture<List<User>> findAllUsers(){
        logger.info("get list of user by " + Thread.currentThread().getName());
        List<User> users = repository.findAll();
        return CompletableFuture.completedFuture(users);
    }

    private List<User> parseCSVFile(final MultipartFile file) throws Exception {
        final List<User> users = new ArrayList<>();
        try {
            try (final BufferedReader br = new BufferedReader(new InputStreamReader(file.getInputStream()))) {
                String line;
                while ((line = br.readLine()) != null) {
                    final String[] data = line.split(",");
                    final User user = new User();
                    user.setName(data[0]);
                    user.setEmail(data[1]);
                    user.setGender(data[2]);
                    users.add(user);
                }
                return users;
            }
        } catch (final IOException e) {
            logger.error("Failed to parse CSV file", e);
            throw new Exception("Failed to parse CSV file {}", e);
        }
    }

}

线程池配置代码 #

package com.mhndev.springexecutor.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.Executor;

@Configuration
@EnableAsync
public class AsyncConfig {

    @Bean(name = "taskExecutor")
    public Executor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(3);
        executor.setMaxPoolSize(3);
        executor.setQueueCapacity(100);
        executor.setThreadNamePrefix("User-Thread-");
        executor.initialize();

        return executor;
    }
}

内容的提问来源于stack exchange，提问作者Majid Abdolhosseini