1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
package cn.gzsendi.utils;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Primary;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;


/**
 * 异步工具类. <br>
 * @see org.springframework.context.annotation.Import
 */
public class AsyncUtils {

    private static final Logger LOGGER = LoggerFactory.getLogger(AsyncUtils.class);

    /**
     * {@link ThreadPoolTaskExecutor}
     * 与{@link Executors#newFixedThreadPool(int)}
     * 内部使用的都是{@link ThreadPoolExecutor}
     */
    private static final ThreadPoolTaskExecutor EXECUTOR;

    static {
        EXECUTOR = new ThreadPoolTaskExecutor() {

            private static final long serialVersionUID = -1644198965440814899L;

            @Override
            protected ExecutorService initializeExecutor(ThreadFactory threadFactory,
                                                         RejectedExecutionHandler rejectedExecutionHandler) {
                try {
                    return this.getThreadPoolExecutor();
                } catch (IllegalStateException ise) {
                    //因为下面已经手动调用了afterPropertiesSet()来初始化线程池，
                    //如果AsyncUtils处于IOC环境中，@Bean("AsyncUtils_TaskExecutor")的使用
                    //会使Spring再次调用afterPropertiesSet()，
                    //所以通过这种方式来避免new出两个java.util.concurrent.ThreadPoolExecutor
                }
                return super.initializeExecutor(threadFactory, rejectedExecutionHandler);
            }
        };

        EXECUTOR.setCorePoolSize(2);
        EXECUTOR.setMaxPoolSize(10);
        EXECUTOR.setAllowCoreThreadTimeOut(true);
        EXECUTOR.setWaitForTasksToCompleteOnShutdown(true);
        EXECUTOR.setThreadNamePrefix("AsyncUtils_");
        EXECUTOR.setTaskDecorator(AsyncUtils::wrapRunnable);
        EXECUTOR.setBeanName("AsyncUtils_TaskExecutor");
        //如果线程池超载了，拒绝策略为: 直接使用调用该execute的线程本身来执行
        EXECUTOR.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        EXECUTOR.afterPropertiesSet();//手动调用afterPropertiesSet()以保证非IOC环境下也可以初始化线程池

        //注意，非IOC环境下需要手动关闭线程池
    }

    /**
     * 获取线程池执行器. <br>
     * 如果想用当前工具类的线程池覆盖TaskExecutionAutoConfiguration的配置 <br>
     * 则可以在SpringBoot启动类或其他配置类上使用@Import({AsyncUtils.class}) <br>
     * 此时，application.properties中spring.task.execution开头的配置将无效 <br>
     * 如果想使用spring.task.execution开头的这些配置，请不要把当前工具类加载到IOC容器中 <br>
     * 或者自己了解TaskExecutionAutoConfiguration和当前工具类的运行过程再另行处理
     *
     * @return ThreadPoolTaskExecutor
     * @see org.springframework.context.annotation.Import
     * <div>@see org.springframework.boot.autoconfigure.task.TaskExecutionProperties</div>
     * <div>@see org.springframework.boot.autoconfigure.task.TaskExecutionAutoConfiguration</div>
     */
    @Bean("AsyncUtils_TaskExecutor")
    @Primary
    public static ThreadPoolTaskExecutor getExecutor() {
        return EXECUTOR;
    }

    /**
     * 关闭线程池. <br>
     * 停止接收外部submit的任务 <br>
     * 内部正在跑的任务和队列里等待的任务，会执行完 <br>
     * 等到第二步完成后，才真正停止
     */
    public static void shutdownExecutor() {
        EXECUTOR.setWaitForTasksToCompleteOnShutdown(true);
        EXECUTOR.shutdown();
    }

    /**
     * 马上关闭线程池. <br>
     * 跟shutdownExecutor()一样，先停止接收外部提交的任务 <br>
     * 忽略队列里等待的任务 <br>
     * 尝试将正在跑的任务interrupt中断 <br>
     * java.util.concurrent.ThreadPoolExecutor的shutdown()将返回未执行的任务列表(见其源码)
     */
    public static void shutdownExecutorNow() {
        EXECUTOR.setWaitForTasksToCompleteOnShutdown(false);
        EXECUTOR.shutdown();
    }

    /**
     * 异步执行任务.
     *
     * @param runnable Runnable的实现类
     */
    public static void asyncExecute(Runnable runnable) {
        if (runnable == null) {
            return;
        }
        EXECUTOR.execute(runnable);
    }

    /**
     * 异步执行任务并获取Future对象.
     *
     * @param callable Callable的实现类
     * @param <T>      泛型类
     * @return Future
     */
    public static <T> Future<T> asyncSubmit(Callable<T> callable) {
        if (callable == null) {
            return null;
        }
        return EXECUTOR.submit(wrapCallable(callable));
    }

    /**
     * 指定线程名开启一个新的异步线程并执行任务. <br>
     * 此方法不使用线程池
     *
     * @param threadName 线程名
     * @param runnable   Runnable的实现类
     */
    public static void asyncExecute(String threadName, Runnable runnable) {
        if (runnable == null) {
            return;
        }
        new Thread(wrapRunnable(runnable), threadName).start();
    }

    /**
     * 包装一下runnable，使得run()异常时捕获异常信息并log一下.
     *
     * @param runnable Runnable实例
     * @return 包装过的Runnable实例
     */
    private static Runnable wrapRunnable(Runnable runnable) {
        return () -> {
            try {
                runnable.run();
            } catch (Throwable t) {
                LOGGER.error("异步线程执行异常 : ", t);
            }
        };
    }

    /**
     * 包装一下runnable，使得call()异常时捕获异常信息并log一下.
     *
     * @param callable Callable实例
     * @return 包装过的Callable实例
     */
    private static <T> Callable<T> wrapCallable(Callable<T> callable) {
        return () -> {
            try {
                return callable.call();
            } catch (Throwable t) {
                LOGGER.error("异步线程执行异常 : ", t);
                throw t;
            }
        };
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
    public static void main(String[] args) throws IOException, InterruptedException {

    	AsyncUtils.asyncExecute(()->{
    		System.out.println("表达式写法执行------");
    	});
    	
    	
    	AsyncUtils.asyncExecute(new Runnable() {
			public void run() {
				System.out.println("普通的写法执行------");
			}
		});
    	
    	System.out.println("1111");
    	
    }

1
2
3
4
表达式写法执行------
1111
普通的写法执行------