ThreadPoolTaskExecutor

快来打我* 2022-07-10 15:58 190阅读 0赞

Spring 擅长对组件的封装和集成， Spring-context对JDK的并发包做了功能增强。

Java代码   ![收藏代码][Image 1]

1.  <bean id="asyncTaskExecutor" **class**="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">  
2.      <property name="corePoolSize" value="10" />  
3.      <property name="maxPoolSize" value="10" />  
4.      <property name="keepAliveSeconds" value="60" />  
5.      <property name="queueCapacity" value="100" />  
6.      <property name="allowCoreThreadTimeOut" value="false" />  
7.  </bean>

以上asyncTaskExecutor,你可以注入到任何一个bean去执行，底层使用JDK的ThreadPoolTaskExecutor来管理线程,默认使用的是JDK的线程池.

Java代码   ![收藏代码][Image 1]

1.  @Autowired  
2.  @Qualifier("asyncTaskExecutor")  
3.  **private** ThreadPoolTaskExecutor asyncTaskExecutor;  
4.    
5.          FutureTask<String> f1 = (FutureTask<String>)asyncTaskExecutor.submit(c1);  
6.          **if** (f1.isDone()) \{  
7.               String result = future.get();  
8.          \}

以上只是简单的应用，非常方便的开发，我们都不用去处理线程池的初始化，以及线程的管理。   
  
Spring 还增加了对线程的监听，这样当现场成功的时候做一些处理，线程失败的时候做一些处理。

Java代码   ![收藏代码][Image 1]

1.  ListenableFutureTask<String> f1 = (ListenableFutureTask<String>) asyncTaskExecutor.submitListenable(c1);  
2.  f1.addCallback(**new** ListenableFutureCallback<String>() \{  
3.    
4.              @Override  
5.              **public** **void** onSuccess(String result) \{  
6.                  //TODO  
7.              \}  
8.    
9.              @Override  
10.             **public** **void** onFailure(Throwable t) \{  
11.                 // TODO Auto-generated method stub  
12.                 t.printStackTrace();  
13.             \}  
14.               
15.            \});

这里Spring非常机智的扩展FutureTask的protected方法

Java代码   ![收藏代码][Image 1]

1.  /\* 
2.   \* Copyright 2002-2013 the original author or authors. 
3.   \* 
4.   \* Licensed under the Apache License, Version 2.0 (the "License"); 
5.   \* you may not use this file except in compliance with the License. 
6.   \* You may obtain a copy of the License at 
7.   \* 
8.   \*      http://www.apache.org/licenses/LICENSE-2.0 
9.   \* 
10.  \* Unless required by applicable law or agreed to in writing, software 
11.  \* distributed under the License is distributed on an "AS IS" BASIS, 
12.  \* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
13.  \* See the License for the specific language governing permissions and 
14.  \* limitations under the License. 
15.  \*/  
16.   
17. **package** org.springframework.util.concurrent;  
18.   
19. **import** java.util.concurrent.Callable;  
20. **import** java.util.concurrent.ExecutionException;  
21. **import** java.util.concurrent.FutureTask;  
22.   
23. /\*\* 
24.  \* Extension of \{@link FutureTask\} that implements \{@link ListenableFuture\}. 
25.  \* 
26.  \* @author Arjen Poutsma 
27.  \* @since 4.0 
28.  \*/  
29. **public** **class** ListenableFutureTask<T> **extends** FutureTask<T> **implements** ListenableFuture<T> \{  
30.   
31.     **private** **final** ListenableFutureCallbackRegistry<T> callbacks = **new** ListenableFutureCallbackRegistry<T>();  
32.   
33.   
34.     /\*\* 
35.      \* Create a new \{@code ListenableFutureTask\} that will, upon running, 
36.      \* execute the given \{@link Callable\}. 
37.      \* @param callable the callable task 
38.      \*/  
39.     **public** ListenableFutureTask(Callable<T> callable) \{  
40.         **super**(callable);  
41.     \}  
42.   
43.     /\*\* 
44.      \* Create a \{@code ListenableFutureTask\} that will, upon running, 
45.      \* execute the given \{@link Runnable\}, and arrange that \{@link \#get()\} 
46.      \* will return the given result on successful completion. 
47.      \* @param runnable the runnable task 
48.      \* @param result the result to return on successful completion 
49.      \*/  
50.     **public** ListenableFutureTask(Runnable runnable, T result) \{  
51.         **super**(runnable, result);  
52.     \}  
53.   
54.   
55.     @Override  
56.     **public** **void** addCallback(ListenableFutureCallback<? **super** T> callback) \{  
57.         **this**.callbacks.addCallback(callback);  
58.     \}  
59.   
60.     @Override  
61.     **protected** **final** **void** done() \{  
62.         Throwable cause;  
63.         **try** \{  
64.             T result = get();  
65.             **this**.callbacks.success(result);  
66.             **return**;  
67.         \}  
68.         **catch** (InterruptedException ex) \{  
69.             Thread.currentThread().interrupt();  
70.             **return**;  
71.         \}  
72.         **catch** (ExecutionException ex) \{  
73.             cause = ex.getCause();  
74.             **if** (cause == **null**) \{  
75.                 cause = ex;  
76.             \}  
77.         \}  
78.         **catch** (Throwable ex) \{  
79.             cause = ex;  
80.         \}  
81.         **this**.callbacks.failure(cause);  
82.     \}  
83.   
84. \}

ListenableFutureCallbackRegistry做了一个简单的策略， 可以添加CallBack, 有两方法success 和 failure.

Java代码   ![收藏代码][Image 1]

1.  /\* 
2.   \* Copyright 2002-2013 the original author or authors. 
3.   \* 
4.   \* Licensed under the Apache License, Version 2.0 (the "License"); 
5.   \* you may not use this file except in compliance with the License. 
6.   \* You may obtain a copy of the License at 
7.   \* 
8.   \*      http://www.apache.org/licenses/LICENSE-2.0 
9.   \* 
10.  \* Unless required by applicable law or agreed to in writing, software 
11.  \* distributed under the License is distributed on an "AS IS" BASIS, 
12.  \* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
13.  \* See the License for the specific language governing permissions and 
14.  \* limitations under the License. 
15.  \*/  
16.   
17. **package** org.springframework.util.concurrent;  
18.   
19. **import** java.util.LinkedList;  
20. **import** java.util.Queue;  
21.   
22. **import** org.springframework.util.Assert;  
23.   
24. /\*\* 
25.  \* Registry for \{@link ListenableFutureCallback\} instances. 
26.  \* 
27.  \* <p>Inspired by \{@code com.google.common.util.concurrent.ExecutionList\}. 
28.  \* 
29.  \* @author Arjen Poutsma 
30.  \* @since 4.0 
31.  \*/  
32. **public** **class** ListenableFutureCallbackRegistry<T> \{  
33.   
34.     **private** **final** Queue<ListenableFutureCallback<? **super** T>> callbacks =  
35.             **new** LinkedList<ListenableFutureCallback<? **super** T>>();  
36.   
37.     **private** State state = State.NEW;  
38.   
39.     **private** Object result = **null**;  
40.   
41.     **private** **final** Object mutex = **new** Object();  
42.   
43.   
44.     /\*\* 
45.      \* Adds the given callback to this registry. 
46.      \* @param callback the callback to add 
47.      \*/  
48.     @SuppressWarnings("unchecked")  
49.     **public** **void** addCallback(ListenableFutureCallback<? **super** T> callback) \{  
50.         Assert.notNull(callback, "'callback' must not be null");  
51.   
52.         **synchronized** (mutex) \{  
53.             **switch** (state) \{  
54.                 **case** NEW:  
55.                     callbacks.add(callback);  
56.                     **break**;  
57.                 **case** SUCCESS:  
58.                     callback.onSuccess((T)result);  
59.                     **break**;  
60.                 **case** FAILURE:  
61.                     callback.onFailure((Throwable) result);  
62.                     **break**;  
63.             \}  
64.         \}  
65.     \}  
66.   
67.     /\*\* 
68.      \* Triggers a \{@link ListenableFutureCallback\#onSuccess(Object)\} call on all added 
69.      \* callbacks with the given result 
70.      \* @param result the result to trigger the callbacks with 
71.      \*/  
72.     **public** **void** success(T result) \{  
73.         **synchronized** (mutex) \{  
74.             state = State.SUCCESS;  
75.             **this**.result = result;  
76.   
77.             **while** (!callbacks.isEmpty()) \{  
78.                 callbacks.poll().onSuccess(result);  
79.             \}  
80.         \}  
81.     \}  
82.   
83.     /\*\* 
84.      \* Triggers a \{@link ListenableFutureCallback\#onFailure(Throwable)\} call on all added 
85.      \* callbacks with the given \{@code Throwable\}. 
86.      \* @param t the exception to trigger the callbacks with 
87.      \*/  
88.     **public** **void** failure(Throwable t) \{  
89.         **synchronized** (mutex) \{  
90.             state = State.FAILURE;  
91.             **this**.result = t;  
92.   
93.             **while** (!callbacks.isEmpty()) \{  
94.                 callbacks.poll().onFailure(t);  
95.             \}  
96.         \}  
97.     \}  
98.   
99.     **private** **enum** State \{NEW, SUCCESS, FAILURE\}  
100.   
101. \}

这里使用LinkedQueue 来维护Callbacks. 使用add()添加,使用pool()方法来获取Queue的first head去执行success/failure,然后删除，使用LinkedQueue对于   
经常insert/delete的队列可是非常快的，很多东西可以学习的(Spring source code).   
  
其实我是不大喜欢自己去实现一套，Spring封装的挺好，跑题了，呵呵.   
ListenableFutureTask 重写了JDK FutureTask.done().

Java代码   ![收藏代码][Image 1]

1.  /\*\* 
2.    \* Protected method invoked when this task transitions to state 
3.    \* \{@code isDone\} (whether normally or via cancellation). The 
4.    \* default implementation does nothing.  Subclasses may override 
5.    \* this method to invoke completion callbacks or perform 
6.    \* bookkeeping. Note that you can query status inside the 
7.    \* implementation of this method to determine whether this task 
8.    \* has been cancelled. 
9.    \*/  
10.  **protected** **void** done() \{ \}

当任务结束的时候，FutureTask会调用这个方法。所以这个方法还能异步调用。   
比如   
   task1  execute 10s   
  task2 execute 20s   
在同时执行的情况下，task1's callback 方法会先返回， 在实际中，如果有10个任务，这个功能先返回执行完成的任务给客户，对于实时性要求比较高的应用，这是个不错的解决方案。

[Image 1]: