springboot使用多线程进行扫描文件以及读取数据库

快来打我* 2024-03-17 10:59 32阅读 0赞

好的，那我就举一个实际业务中使用@Async注解的例子吧。

假设我们有一个需求，需要在用户发表一篇文章时，对这篇文章进行一个异步的处理，比如将文章的内容进行敏感词过滤。而此时我们又不希望用户等待这个过程完成才能得到反馈，因此我们可以使用@Async注解来异步处理这个任务，让用户不必等待。

1.  首先需要在启动类上添加@EnableAsync注解

@SpringBootApplication
    @EnableAsync
    public class Application {
        public static void main(String[] args) {
            SpringApplication.run(Application.class, args);
        }
    }

1.  在Service层的方法中使用@Async注解来异步执行任务

@Service
    public class ArticleService {
        
        @Async
        public void filterSensitiveWords(Article article){
            //todo:过滤文章敏感词
        }
        
        public void addArticle(Article article) {
            //添加文章
            //异步执行敏感词过滤
            filterSensitiveWords(article);
        }
    }

在上面的代码中，我们将添加文章的方法addArticle中的任务filterSensitiveWords使用@Async注解进行了异步处理，这样当用户调用addArticle方法时，敏感词过滤任务会异步执行，用户不必等待它的完成。

再举一个例子，对查询数据库的大文件进行处理，是不是也用多线程

是的，处理大文件时使用多线程可以提高文件处理的效率，让处理速度更快。以下是使用@Async注解来实现对文件内容的处理的示例代码。

假设我们有一个需求，需要从一个非常大的csv文件中读取数据，然后将其中某些字段的值进行处理，最后将处理后的数据保存到数据库中。由于读取大文件和处理数据都需要耗费时间，因此我们可以使用多线程提高文件处理速度。具体实现如下：

1.  在启动类上添加@EnableAsync注解

@SpringBootApplication
    @EnableAsync
    public class Application {
        public static void main(String[] args) {
            SpringApplication.run(Application.class, args);
        }
    }

1.  定义文件处理任务

@Component
    public class FileProcessTask {
        
        @Autowired
        private DataProcessor dataProcessor;
        
        @Async
        public void process(File file){
            try {
                BufferedReader br = new BufferedReader(new FileReader(file));
                String line;
                while ((line = br.readLine()) != null) {
                    // 对每一行数据进行处理
                    String[] values = line.split(",");
                    for(int i=0;i<values.length;i++){
                        values[i] = dataProcessor.process(values[i]);
                    }
                    // 处理完后保存到数据库中
                    saveToDb(values);
                }
                br.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    
        public void saveToDb(String[] values){
            // 将处理后的数据存储到数据库中
        }
    }

在上面的代码中，我们通过@Async注解将文件处理任务process方法定义为异步执行。在这个方法中，我们使用BufferedReader从文件中读取每一行数据，然后对数据的每一个字段进行处理，处理后将数据保存到数据库。由于这是一个耗时的操作，因此我们使用@Async注解将处理逻辑放在一个独立的线程中执行，这样可以让主线程继续执行其他任务。

1.  在主线程中调用异步方法开始文件处理

@Service
    public class FileService {
        
        @Autowired
        private FileProcessTask fileProcessTask;
        
        public void process(File file) {
            // 调用异步方法
            fileProcessTask.process(file);
        }
    }

在上面的代码中，我们将文件处理任务注入到FileService中，然后在process方法中调用异步方法fileProcessTask.process来开始文件处理过程。由于@Async注解的作用，文件处理任务会在一个独立的线程中执行，而主线程可以继续处理其他任务。

确实，在多线程环境下，为避免重复读取数据行，我们需要进行同步处理，可以使用`java.util.concurrent.ConcurrentHashMap`来记录每个线程已经读取到的行数。具体实现如下：

1.  修改文件处理任务，记录每个线程已经处理的行数

@Component
    public class FileProcessTask {
        
        @Autowired
        private DataProcessor dataProcessor;
    
        // 使用ConcurrentHashMap记录每个线程已读取的行数
        private ConcurrentHashMap<Long, Integer> lineIndexMap = new ConcurrentHashMap<>();
    
        @Async
        public void process(File file){
            try {
                Long threadId = Thread.currentThread().getId();
                BufferedReader br = new BufferedReader(new FileReader(file));
                String line;
                int index = lineIndexMap.getOrDefault(threadId, 0);
                for(int i = 0; i < index; i++) {
                    br.readLine();
                }
                while ((line = br.readLine()) != null) {
                    // 对每一行数据进行处理
                    String[] values = line.split(",");
                    for(int i=0;i<values.length;i++){
                        values[i] = dataProcessor.process(values[i]);
                    }
                    // 处理完后保存到数据库中
                    saveToDb(values);
                    // 更新已读取的行数
                    lineIndexMap.put(threadId, index++);
                }
                br.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    
        public void saveToDb(String[] values){
            // 将处理后的数据存储到数据库中
        }
    }

在上面的代码中，我们增加了一个`ConcurrentHashMap<Long, Integer>`类型的成员变量`lineIndexMap`，来记录每个线程已经处理的行数。在处理任务时，首先获取当前线程的ID `threadId`，然后根据`lineIndexMap`获取当前线程已经处理的行数`index`，从文件读取的位置开始向下处理，处理完一行数据后将已读取的行数更新到`lineIndexMap`中。由于`ConcurrentHashMap`是线程安全的，因此不会出现线程安全问题。

1.  修改启动类，开启线程池支持

@SpringBootApplication
    @EnableAsync
    public class Application {
        
        @Value("${file.path}")
        private String filePath;
    
        public static void main(String[] args) {
            SpringApplication.run(Application.class, args);
        }
    
        @Bean(name = "threadPoolTaskExecutor")
        public ThreadPoolTaskExecutor createThreadPoolTaskExecutor() {
            ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
            executor.setCorePoolSize(10);
            executor.setMaxPoolSize(20);
            executor.setQueueCapacity(100);
            executor.setThreadNamePrefix("FileProcess-");
            executor.initialize();
            return executor;
        }
    
        @Bean
        public CommandLineRunner commandLineRunner() {
            return new CommandLineRunner() {
                @Autowired
                private FileProcessTask fileProcessTask;
    
                @Override
                public void run(String... args) throws Exception {
                    File file = new File(filePath);
                    int threadNum = 10;
                    for (int i = 0; i < threadNum; i++) {
                        // 将文件处理任务提交到线程池中执行
                        createThreadPoolTaskExecutor().submit(() -> fileProcessTask.process(file));
                    }
                }
            };
        }
    }

在上面的代码中，我们增加了一个名为`threadPoolTaskExecutor`的@Bean，来创建一个线程池。在CommandLineRunner的实现中，我们将文件处理任务提交到线程池中执行，确保多个线程能够同时进行文件处理任务。由于线程池中线程数目已经限制在`CorePoolSize`和`MaxPoolSize`范围内，因此不会导致系统资源的浪费。

`CommandLineRunner`接口是Spring Boot提供的一种方式，允许我们在Spring Boot应用程序启动后运行自定义的逻辑。当Spring Boot应用程序启动完成后，所有的CommandLineRunner都会被自动调用。这样做的好处是我们可以进行一些初始化操作，清理内存或者启动其他的线程运行某些任务等等。所以我们将这个bean放到主类中是为了在Spring Boot应用程序启动后就执行一个线程池任务，从而异步地处理文件中的数据行。

/\*\*  
\* 异步处理文件  
\* @param file 待处理的文件  
\*/  
@Async  
public void process(File file) \{  
try \{  
Long threadId = Thread.currentThread().getId();  
BufferedReader br = new BufferedReader(new FileReader(file));  
String line;  
int index = lineIndexMap.getOrDefault(threadId, 0); // 获取当前线程上一次已经处理的行数，默认为0  
for (int i = 0; i < index; i++) \{ // 跳过之前已经处理过的行  
br.readLine();  
\}  
while ((line = br.readLine()) != null) \{ // 循环处理文件中的每一行数据  
// 对每一行数据进行处理，以逗号分隔并保存到一个数组中  
String\[\] values = line.split(",");  
for (int i = 0; i < values.length; i++) \{ // 根据业务逻辑对数据进行处理  
values\[i\] = dataProcessor.process(values\[i\]);  
\}  
// 处理完后保存到数据库中  
saveToDb(values);  
// 更新当前线程已读取的行数  
lineIndexMap.put(threadId, index++);  
\}  
br.close();  
\} catch (Exception e) \{  
e.printStackTrace();  
\}  
\}