SpringCloud（23）—— Spring Boot使用@Async實現異步調用：ThreadPoolTaskScheduler線程池的優雅關閉

定義了一個線程池，然后利用 @Async注解寫了3個任務，並指定了這些任務執行使用的線程池

1、我們定義一個 ThreadPoolTaskScheduler線程池

package com.sinosoft.config;

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

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

/**
 * Created by xushuyi on 2018/4/2.
 */
@EnableAsync
@Configuration
public class ThreadPoolConfig {

    @Bean("taskExecutor")
    public Executor taskExecutor() {
        // 創建一個線程池對象
        ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
        // 定義一個線程池大小
        scheduler.setPoolSize(100);
        // 線程池名的前綴
        scheduler.setThreadNamePrefix("taskExecutor-");
        // 設置線程池關閉的時候等待所有任務都完成再繼續銷毀其他的Bean
        scheduler.setWaitForTasksToCompleteOnShutdown(true);
        // 設置線程池中任務的等待時間，如果超過這個時候還沒有銷毀就強制銷毀，以確保應用最后能夠被關閉，而不是阻塞住
        scheduler.setAwaitTerminationSeconds(60);
        // 線程池對拒絕任務的處理策略,當線程池沒有處理能力的時候，該策略會直接在 execute 方法的調用線程中運行被拒絕的任務；如果執行程序已關閉，則會丟棄該任務
        scheduler.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        return scheduler;
    }
}

2、建立異步任務、讓它依賴一個外部資源，比如：Redis

package com.sinosoft.common;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * Created by xushuyi on 2018/4/2.
 * 查看redis api：
 * https://docs.spring.io/spring-data/data-keyvalue/docs/current/api/org/springframework/data/keyvalue/redis/core/StringRedisTemplate.html
 */
@Slf4j
@Component
public class TaskScheduler {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Async("taskExecutor")
    public void doTaskOne() {
        log.info("開始做任務一...");
        long start = System.currentTimeMillis();
        log.info(stringRedisTemplate.toString());

        // 向Redis寫入數據 永久性
        stringRedisTemplate.opsForValue().set("username", "xushuyi");

        // 刪除Redis中的Key值
        stringRedisTemplate.delete("username1");

        // 向Redis寫入數據 + 緩存時間 1分鍾
        stringRedisTemplate.opsForValue().set("username2", "zhangsan", 60 * 1, TimeUnit.SECONDS);

        // 隨機獲取Redis中的Key值
        log.info("獲取隨機Redis Key值：" + stringRedisTemplate.randomKey());

        // 向Redis寫入num 數值
        stringRedisTemplate.opsForValue().set("num", "100");

        // 獲取Redis的Key num對應的值
        log.info("獲取寫入Redis num的值：" + stringRedisTemplate.opsForValue().get("num"));

        // 對Redis中的num數值進行減操作
        stringRedisTemplate.boundValueOps("num").increment(-1);// val 做 -1 操作

        // 獲取Redis的Key num 進行減操作后對應的值
        log.info("獲取寫入Redis num減去1操作的值：" + stringRedisTemplate.opsForValue().get("num"));

        // 獲取Redis的Key num 進行加操作
        stringRedisTemplate.boundValueOps("num").increment(2);// val 做 +1 操作

        // 獲取Redis的Key num 進行加操作后對應的值
        log.info("獲取寫入Redis num加上2操作的值：" + stringRedisTemplate.opsForValue().get("num"));

        // 獲取Redis Key值的過期時間
        log.info("獲取寫入Redis username的過期時間" + stringRedisTemplate.getExpire("username"));
        log.info("獲取寫入Redis username2的過期時間" + stringRedisTemplate.getExpire("username2"));

        // 獲取Redis Key值過期時間 並換算為指定時間單位 例如：天
        log.info("獲取寫入Redis username的過期時間" + stringRedisTemplate.getExpire("username", TimeUnit.SECONDS));
        log.info("獲取寫入Redis username2的過期時間" + stringRedisTemplate.getExpire("username2", TimeUnit.SECONDS));

        // 判斷Redis是否包含對應的Key值
        log.info("判斷Redis 是否包含Key username：" + stringRedisTemplate.hasKey("username"));
        log.info("判斷Redis 是否包含Key username1：" + stringRedisTemplate.hasKey("username1"));

        // 向Redis 指定Key放入對象或集合
        stringRedisTemplate.opsForSet().add("set_123", "1", "2", "3");
        stringRedisTemplate.expire("set_123", 1000, TimeUnit.MILLISECONDS);//設置過期時間
        log.info("根據key查看集合中是否存在指定數據：" + stringRedisTemplate.opsForSet().isMember("set_123", "1"));
        log.info("查看集合數據：" + stringRedisTemplate.opsForSet().members("set_123"));

        // 通過Redis Key值獲取對應的有效時間
        Long expire = stringRedisTemplate.boundHashOps("set_123").getExpire();
        log.info("集合有效時間：" + expire + "S");

        // 向Redis 中寫入Map對象 其中 三個參數 為：redis key，對應map 的 key 及 val
        User user = new User("xushuyi", "123");
        stringRedisTemplate.opsForHash().putIfAbsent("key1", "xushuyi", user.toString());

        // 獲取Redis中的Map對象
        Map backUser = stringRedisTemplate.opsForHash().entries("key1");
        log.info("獲取Redis中的Map對象：" + backUser);
        log.info("獲取Redis Key值下的某一個Map對象：" + stringRedisTemplate.opsForHash().get("key1", "xushuyi"));


        long end = System.currentTimeMillis();
        log.info("完成任務一,共計耗時：" + (end - start) + "毫秒.");
    }

    @Async("taskExecutor")
    public void doTaskTwo() {
        log.info("開始做任務二...");
        long start = System.currentTimeMillis();
        log.info(stringRedisTemplate.randomKey());
        long end = System.currentTimeMillis();
        log.info("完成任務二,共計耗時：" + (end - start) + "毫秒.");
    }

    @Async("taskExecutor")
    public void doTaskThree() {
        log.info("開始做任務三...");
        long start = System.currentTimeMillis();
        log.info(stringRedisTemplate.randomKey());
        long end = System.currentTimeMillis();
        log.info("完成任務三,共計耗時：" + (end - start) + "毫秒.");
    }

}

@Data
class User {
    private String username;
    private String pwd;

    public User(String username, String pwd) {
        this.username = username;
        this.pwd = pwd;
    }

}

3、pom.xml 中配置

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>micro</artifactId>
        <groupId>com.sinosoft</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>threadPool-task</artifactId>
    <packaging>jar</packaging>

    <dependencies>

        <!-- 添加Eureka的依賴 -->
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-eureka</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-feign</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-jpa</artifactId>
        </dependency>

        <dependency>
            <groupId>com.h2database</groupId>
            <artifactId>h2</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>

        <!-- Swagger UI -->
        <dependency>
            <groupId>io.springfox</groupId>
            <artifactId>springfox-swagger2</artifactId>
            <version>${springfox.version}</version>
        </dependency>
        <dependency>
            <groupId>io.springfox</groupId>
            <artifactId>springfox-swagger-ui</artifactId>
            <version>${springfox.version}</version>
        </dependency>

        <dependency>
            <groupId>com.google.guava</groupId>
            <artifactId>guava</artifactId>
            <version>19.0</version>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-redis</artifactId>
        </dependency>

        <!-- 整合hystrix，其實feign中自帶了hystrix，引入該依賴主要是為了使用其中的hystrix-metrics-event-stream，用於dashboard -->
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-hystrix</artifactId>
        </dependency>
        <dependency>
            <groupId>com.sinosoft</groupId>
            <artifactId>common</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>
        <dependency>
            <groupId>com.sinosoft</groupId>
            <artifactId>interface</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>

4、修改單元測試 模擬高並發shutdown情況

package com.sinosoft;

import com.sinosoft.common.TaskScheduler;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.context.annotation.Import;
import org.springframework.test.context.ActiveProfiles;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.web.WebAppConfiguration;

/**
 * Created by xushuyi on 2018/4/2.
 */
@Slf4j
@RunWith(SpringJUnit4ClassRunner.class)
@SpringBootTest
@Import(ThreadApplication.class)
@ActiveProfiles(profiles = "")
@WebAppConfiguration
public class TaskSchedulerTest {


    @Autowired
    private TaskScheduler taskScheduler;


    @Before
    public void before() {
        log.info("開始單元測試...");
    }

    @Test
    @SneakyThrows
    public void taskTest() {
        taskScheduler.doTaskOne();
        for (int i = 0; i < 100; i++) {
            taskScheduler.doTaskTwo();
            taskScheduler.doTaskThree();
            // 模擬高並發 情況下 系統 shutdown
            if (i == 50) {
                System.exit(0);
            }
        }
    }

    @After
    public void after() {
        log.info("結束單元測試...");
    }

}