Springboot多线程配置方法

package test;

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("test")
@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;
  }
}

package test;

import java.util.Random;
import java.util.concurrent.Future;

import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.stereotype.Service;

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

  Random random = new Random();// 默认构造方法

  @Async
  // 表明是异步方法
  // 无返回值
  public void executeAsyncTask(Integer i) {
    System.out.println("执行异步任务：" + i);
  }

  /**
   * 异常调用返回Future
   * 
   * @param i
   * @return
   * @throws InterruptedException
   */
  @Async
  public Future asyncInvokeReturnFuture(int i) throws InterruptedException {
    System.out.println("input is " + i);
    Thread.sleep(1000 * random.nextInt(i));

    Future future = new AsyncResult("success:" + i);// Future接收返回值，这里是String类型，可以指明其他类型

    return future;
  }
}

package test;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.core.task.TaskRejectedException;

public class Application {

  public static void main(String[] args) throws InterruptedException, ExecutionException {
    // testVoid();

    testReturn();
  }

  // 测试无返回结果
  private static void testVoid() {
    AnnotationConfigApplicationContext cOntext= new AnnotationConfigApplicationContext(AsyncTaskConfig.class);
    AsyncTaskService asyncTaskService = context.getBean(AsyncTaskService.class);

    // 创建了20个线程
    for (int i = 1; i <= 20; i++) {
      asyncTaskService.executeAsyncTask(i);
    }

    context.close();
  }

  // 测试有返回结果
  private static void testReturn() throws InterruptedException, ExecutionException {
    AnnotationConfigApplicationContext cOntext= new AnnotationConfigApplicationContext(AsyncTaskConfig.class);
    AsyncTaskService asyncTaskService = context.getBean(AsyncTaskService.class);

    List> lstFuture = new ArrayList>();// 存放所有的线程，用于获取结果

    // 创建100个线程
    for (int i = 1; i <= 100; i++) {
      while (true) {
        try {
          // 线程池超过最大线程数时，会抛出TaskRejectedException，则等待1s，直到不抛出异常为止
          Future future = asyncTaskService.asyncInvokeReturnFuture(i);
          lstFuture.add(future);

          break;
        } catch (TaskRejectedException e) {
          System.out.println("线程池满，等待1S。");
          Thread.sleep(1000);
        }
      }
    }

    // 获取值。get是阻塞式，等待当前线程完成才返回值
    for (Future future : lstFuture) {
      System.out.println(future.get());
    }

    context.close();
  }
}

 4.0.0
 TestAysc
 TestAysc
 0.0.1-SNAPSHOT
 
   
     org.springframework.boot
     spring-boot
     1.5.6.RELEASE
   
   
     org.springframework
     spring-aop
     4.3.10.RELEASE