線程池創(chuàng)建類

package org.pzy.spring.complex;

import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;

import org.springframework.beans.factory.annotation.Configurable;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import lombok.Data;

@Component
@Configurable
@ConfigurationProperties(prefix = "myThreadPool")
@Data
public class ThreadPoolConfiguration {

  private int corePoolSize;
  private int maxPoolSize;
  private int queueCapacity;
  private int keepAliveSeconds;

  @Bean
  public Executor testMyThreadPool01() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(corePoolSize);
    executor.setMaxPoolSize(maxPoolSize);
    executor.setQueueCapacity(queueCapacity);
    executor.setKeepAliveSeconds(keepAliveSeconds);
    executor.setThreadNamePrefix("my-thread-pool-01");
    // 當(dāng)線程池已滿,且等待隊(duì)列也滿了的時(shí)候,轉(zhuǎn)為主線程執(zhí)行
    executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
    executor.initialize();
    return executor;
  }

  @Bean
  public Executor testMyThreadPool02() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(corePoolSize);
    executor.setMaxPoolSize(maxPoolSize);
    executor.setQueueCapacity(queueCapacity);
    executor.setKeepAliveSeconds(keepAliveSeconds);
    executor.setThreadNamePrefix("my-thread-pool-02");
    // 當(dāng)線程池已滿,且等待隊(duì)列也滿了的時(shí)候,拋出TaskRejectedException
    executor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
    executor.initialize();
    return executor;
  }

  @Bean
  public Executor testMyThreadPool03() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(corePoolSize);
    executor.setMaxPoolSize(maxPoolSize);
    executor.setQueueCapacity(queueCapacity);
    executor.setKeepAliveSeconds(keepAliveSeconds);
    executor.setThreadNamePrefix("my-thread-pool-03");
    // 當(dāng)線程池已滿,且等待隊(duì)列也滿了的時(shí)候,拋棄一個(gè)在等待隊(duì)列中等待的時(shí)間最久的線程,并將當(dāng)前線程放入等待隊(duì)列(不會(huì)拋出異常)
    executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy());
    executor.initialize();
    return executor;
  }

  @Bean
  public Executor testMyThreadPool04() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(corePoolSize);
    executor.setMaxPoolSize(maxPoolSize);
    executor.setQueueCapacity(queueCapacity);
    executor.setKeepAliveSeconds(keepAliveSeconds);
    executor.setThreadNamePrefix("my-thread-pool-04");
    // 當(dāng)線程池已滿,且等待隊(duì)列也滿了的時(shí)候,直接拋棄當(dāng)前線程(不會(huì)拋出異常)
    executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy());
    executor.initialize();
    return executor;
  }
}

使用線程池

package org.pzy.spring.complex.bean;

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

@Service
public class TestService {

  @Async("testMyThreadPool01")
  public void say(int id) throws InterruptedException {
    System.out.println(Thread.currentThread().getName() + " " + Thread.currentThread().getId() + " --> " + id);
    Thread.sleep(2 * 1000);
  }

  @Async("testMyThreadPool02")
  public void say02(int id) throws InterruptedException {
    System.out.println(Thread.currentThread().getName() + " " + Thread.currentThread().getId() + " --> " + id);
    Thread.sleep(2 * 1000);
  }

  @Async("testMyThreadPool03")
  public void say03(int id) throws InterruptedException {
    System.out.println(Thread.currentThread().getName() + " " + Thread.currentThread().getId() + " --> " + id);
    Thread.sleep(2 * 1000);
  }

  @Async("testMyThreadPool04")
  public void say04(int id) throws InterruptedException {
    System.out.println(Thread.currentThread().getName() + " " + Thread.currentThread().getId() + " --> " + id);
    Thread.sleep(2 * 1000);
  }
}

測(cè)試類

package org.pzy.spring.complex;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.pzy.spring.complex.bean.TestService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

@SpringBootTest
@RunWith(SpringJUnit4ClassRunner.class)
public class AppTest {

  @Autowired
  private TestService testService;

  @Test
  public void test() throws InterruptedException {
    for (int i = 0; i < 10; i++) {
      testService.say(i);
    }
    Thread.sleep(10000 * 1000);
  }

  @Test
  public void test02() throws InterruptedException {
    for (int i = 0; i < 10; i++) {
      testService.say02(i);
    }
    Thread.sleep(10000 * 1000);
  }
  
  @Test
  public void test03() throws InterruptedException {
    for (int i = 0; i < 10; i++) {
      testService.say03(i);
    }
    Thread.sleep(10000 * 1000);
  }
  
  @Test
  public void test04() throws InterruptedException {
    for (int i = 0; i < 10; i++) {
      testService.say04(i);
    }
    Thread.sleep(10000 * 1000);
  }
}

配置文件

server:
  port: 56000
myThreadPool:
  corePoolSize: 1   # 核心線程池大小
  maxPoolSize: 3    # 最大線程池大小
  queueCapacity: 2  # 等待隊(duì)列大小
  keepAliveSeconds: 3 # 空閑線程存活時(shí)間

啟動(dòng)類

package org.pzy.spring.complex;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableAsync;

@SpringBootApplication
@EnableAsync //啟用異步編程
public class App {
  public static void main(String[] args) {
    SpringApplication.run(App.class);
  }
}

代碼詳見(jiàn): https://gitee.com/free_pan/spring-summary/tree/master/spring-complex-04