SpringBoot 多线程

Posted 2021-01-15 miye

Spring通过任务执行器（TaskExecutor）来实现多线程和并发编程。使用ThreadPoolTaskExecutor可实现一个基于线程池的TaskExecutor。而实际开发中任务一般是非阻碍的，即异步的，所以我们要在配置类中通过@EnableAsync 开启对异步任务的支持，并通过实际执行Bean的方法中使用@Async注解来声明其是一个异步任务。

示例：

1.配置类。

package com.cnpiec.ireader.config;

import java.util.concurrent.Executor;

import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.AsyncConfigurer;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

@Configuration
@ComponentScan("com.cnpiec.ireader")
@EnableAsync
// 线程配置类
public class AsyncTaskConfig implements AsyncConfigurer {

    // ThredPoolTaskExcutor的处理流程
    // 当池子大小小于corePoolSize，就新建线程，并处理请求
    // 当池子大小等于corePoolSize，把请求放入workQueue中，池子里的空闲线程就去workQueue中取任务并处理
    // 当workQueue放不下任务时，就新建线程入池，并处理请求，如果池子大小撑到了maximumPoolSize，就用RejectedExecutionHandler来做拒绝处理
    // 当池子的线程数大于corePoolSize时，多余的线程会等待keepAliveTime长时间，如果无请求可处理就自行销毁

    @Override
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        taskExecutor.setCorePoolSize(5);// 最小线程数
        taskExecutor.setMaxPoolSize(10);// 最大线程数
        taskExecutor.setQueueCapacity(25);// 等待队列

        taskExecutor.initialize();

        return taskExecutor;
    }

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return null;
    }
}

2.任务执行类

package com.cnpiec.ireader.service;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.util.List;
import java.util.concurrent.Future;

import org.apache.http.HttpEntity;
import org.apache.http.StatusLine;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.stereotype.Service;
import org.springframework.util.DigestUtils;

import com.alibaba.fastjson.JSONObject;
import com.cnpiec.ireader.dao.GetBookDataDao;
import com.cnpiec.ireader.model.Book;
import com.cnpiec.ireader.model.Chapter;
import com.cnpiec.ireader.utils.FileNameUtils;

@Service
// 线程执行任务类
public class AsyncDownloadBookTaskService {

    private Logger logger = LoggerFactory.getLogger(AsyncDownloadBookTaskService.class);

    @Autowired
    private GetBookDataDao getBookDataDao;

    @Async
    // 表明是异步方法
    // 无返回值
    public void executeAsyncTask(List<Book> list, String clientId, String clientSecret) {
        System.out.println(Thread.currentThread().getName() + "开启新线程执行");
        for (Book book : list) {
            String name = book.getName();
            File file = new File("iReaderResource/" + name);
            if (file.mkdirs()) {
                logger.info("文件夹创建成功！创建后的文件目录为：" + file.getPath());
            }
            try {
                getChapterList(book, clientId, clientSecret);
            } catch (Exception e) {
                logger.error("多线程下载书籍失败:" + book.getBookId(), e);
                e.printStackTrace();
            }
        }
    }

    /**
     * 异常调用返回Future
     *
     * @param i
     * @return
     * @throws InterruptedException
     */
    @Async
    public Future<String> asyncInvokeReturnFuture(List<Book> list, String clientId, String clientSecret)
            throws InterruptedException {

        //业务代码
        Future<String> future = new AsyncResult<String>("success");// Future接收返回值，这里是String类型，可以指明其他类型

        return future;
    }

    /**
     * 根据书籍ID获取章节创建文件并写入章节信息
     *
     * @param book
     * @param clientId
     * @param clientSecret
     * @throws Exception
     */
    private void getChapterList(Book book, String clientId, String clientSecret) throws Exception {
        String stempSign = clientId + clientSecret + book.getBookId();
        String sign = DigestUtils.md5DigestAsHex(stempSign.getBytes());
        CloseableHttpClient httpclient = HttpClients.createDefault();
        StringBuffer sb = new StringBuffer();
        sb.append("http://api.res.ireader.com/api/v2/book/chapterList?").append("bookId=").append(book.getBookId())
                .append("&clientId=").append(clientId).append("&sign=").append(sign).append("&resType=json");
        HttpGet httpget = new HttpGet(sb.toString());
        CloseableHttpResponse response = httpclient.execute(httpget);

        StatusLine statusLine = response.getStatusLine();
        if (statusLine.getStatusCode() == 200) {
            HttpEntity entity = response.getEntity();
            String result = EntityUtils.toString(entity, "utf-8");
            List<Chapter> chapterList = JSONObject.parseArray(result, Chapter.class);
            for (Chapter chapter : chapterList) {
                File file = new File("iReaderResource/" + book.getName() + "/"
                        + FileNameUtils.replaceSpecialCharacters(chapter.getTitle()) + ".txt");
                if (file.createNewFile()) {
                    logger.info("创建章节文件成功：" + file.getPath());
                }
                String filePath = file.getPath();
                getChapterInfo(chapter, book, clientId, clientSecret, filePath);

            }
            getBookDataDao.updateBookStatus(book.getBookId());
        }

    }

    /**
     * 获取章节信息写入文本文件
     *
     * @param chapter
     * @param book
     * @param clientId
     * @param clientSecret
     * @param filePath
     * @throws Exception
     */
    private void getChapterInfo(Chapter chapter, Book book, String clientId, String clientSecret, String filePath)
            throws Exception {
        String stempSign = clientId + clientSecret + book.getBookId() + chapter.getChapterId();
        String sign = DigestUtils.md5DigestAsHex(stempSign.getBytes());
        CloseableHttpClient httpclient = HttpClients.createDefault();
        StringBuffer sb = new StringBuffer();
        sb.append("http://api.res.ireader.com/api/v2/book/chapterInfo?").append("bookId=").append(book.getBookId())
                .append("&chapterId=").append(chapter.getChapterId()).append("&clientId=").append(clientId)
                .append("&sign=").append(sign).append("&resType=json");
        HttpGet httpget = new HttpGet(sb.toString());
        CloseableHttpResponse response = httpclient.execute(httpget);

        StatusLine statusLine = response.getStatusLine();
        if (statusLine.getStatusCode() == 200) {
            HttpEntity entity = response.getEntity();
            String result = EntityUtils.toString(entity, "utf-8");
            Chapter chapter2 = JSONObject.parseObject(result, Chapter.class);
            String content = chapter2.getContent();
            // 写文件内容
            BufferedWriter writer = new BufferedWriter(new FileWriter(new File(filePath), true));
            writer.write(content);
            writer.close();
        }
    }
}

3.运行

for (int i = 0; i < list2.size(); i++) {
     asyncTaskService.executeAsyncTask(list2.get(i),clientId,clientSecret);
 }