线程池的应用及Callable接口的使用

Java代码  

1. public interface Executor {
2. /**
3. * Executes the given command at some time in the future.  The command
4. * may execute in a new thread, in a pooled thread, or in the calling
5. * thread, at the discretion of the <tt>Executor</tt> implementation.
6. *
7. * @param command the runnable task
8. * @throws RejectedExecutionException if this task cannot be
9. * accepted for execution.
10. * @throws NullPointerException if command is null
11. */
12. void execute(Runnable command);
13. }

public interface Executor {

    /**
     * Executes the given command at some time in the future.  The command
     * may execute in a new thread, in a pooled thread, or in the calling
     * thread, at the discretion of the <tt>Executor</tt> implementation.
     *
     * @param command the runnable task
     * @throws RejectedExecutionException if this task cannot be
     * accepted for execution.
     * @throws NullPointerException if command is null
     */
    void execute(Runnable command);
}

Java代码  

1. //接口ExecutorService继承自Executor，它的目的是为我们管理Thread对象，从而简化并发编程
2. public interface ExecutorService extends Executor {
3. <T> Future<T> submit(Callable<T> task);
4. <T> Future<T> submit(Runnable task, T result);
5. Future<?> submit(Runnable task);
6. ...
7. }

//接口ExecutorService继承自Executor，它的目的是为我们管理Thread对象，从而简化并发编程
public interface ExecutorService extends Executor {

    <T> Future<T> submit(Callable<T> task);

    <T> Future<T> submit(Runnable task, T result);

    Future<?> submit(Runnable task);

    ...
}

Java代码  

1. public interface Callable<V> {
2. /**
3. * Computes a result, or throws an exception if unable to do so.
4. *
5. * @return computed result
6. * @throws Exception if unable to compute a result
7. */
8. V call() throws Exception;
9. }
10. public interface Runnable {
11. public abstract void run();
12. }

public interface Callable<V> {
    /**
     * Computes a result, or throws an exception if unable to do so.
     *
     * @return computed result
     * @throws Exception if unable to compute a result
     */
    V call() throws Exception;
}

public interface Runnable {

    public abstract void run();
}

Java代码  

1. public interface Future<V> {
2. boolean cancel(boolean mayInterruptIfRunning);
3. /**
4. * Waits if necessary for the computation to complete, and then
5. * retrieves its result.
6. *
7. * @return the computed result
8. */
9. V get() throws InterruptedException, ExecutionException;
10. V get(long timeout, TimeUnit unit)
11. throws InterruptedException, ExecutionException, TimeoutException;
12. }

public interface Future<V> {

    boolean cancel(boolean mayInterruptIfRunning);    

    /**
     * Waits if necessary for the computation to complete, and then
     * retrieves its result.
     *
     * @return the computed result
     */
    V get() throws InterruptedException, ExecutionException;

    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

Callable接口和Runnable接口相似，区别就是Callable需要实现call方法，而Runnable需要实现run方法；并且，call方法还可以返回任何对象，无论是什么对象，JVM都会当作Object来处理。但是如果使用了泛型，我们就不用每次都对Object进行转换了。

Runnable和Callable都是接口

不同之处：
1.Callable可以返回一个类型V，而Runnable不可以
2.Callable能够抛出checked exception,而Runnable不可以。
3.Runnable是自从java1.1就有了，而Callable是1.5之后才加上去的
4.Callable和Runnable都可以应用于executors。而Thread类只支持Runnable.
上面只是简单的不同，其实这两个接口在用起来差别还是很大的。Callable与executors联合在一起，在任务完成时可立刻获得一个更新了的Future。而Runable却要自己处理

Future接口，一般都是取回Callable执行的状态用的。其中的主要方法：

• cancel，取消Callable的执行，当Callable还没有完成时
• get，获得Callable的返回值
• isCanceled，判断是否取消了
• isDone，判断是否完成

用Executor来构建线程池，应该要做的事：

1).调用Executors类中的静态方法newCachedThreadPool(必要时创建新线程，空闲线程会被保留60秒)或newFixedThreadPool(包含固定数量的线程池)等，返回的是一个实现了ExecutorService接口的ThreadPoolExecutor类或者是一个实现了ScheduledExecutorServiece接口的类对象。

2).调用submit提交Runnable或Callable对象。

3).如果想要取消一个任务，或如果提交Callable对象，那就要保存好返回的Future对象。

4).当不再提交任何任务时，调用shutdown方法。

举2个例子如下：

Java代码   

1. package thread.test04;
2. import java.util.concurrent.*;
3. public class ThreadTestA {
4. public static void main(String[] args) {
5. ExecutorService e=Executors.newFixedThreadPool(10);
6. e.execute(new MyRunnableA());
7. e.execute(new MyRunnableB());
8. e.shutdown();
9. }
10. }
11. class MyRunnableA implements Runnable{
12. public void run(){
13. System.out.println("Runnable:run()....");
14. int i=0;
15. while(i<20){
16. i++;
17. for(int j=0;j<1000000;j++);
18. System.out.println("i="+i);
19. }
20. }
21. }
22. class MyRunnableB implements Runnable{
23. public void run(){
24. char c=‘A‘-1;
25. while(c<‘Z‘){
26. c++;
27. for(int j=0;j<1000000;j++);
28. System.out.println("c="+c);
29. }
30. }
31. }

Java代码   

1. package thread.test04;
2. import java.util.concurrent.Callable;
3. import java.util.concurrent.ExecutionException;
4. import java.util.concurrent.ExecutorService;
5. import java.util.concurrent.Executors;
6. import java.util.concurrent.Future;
7. public class ThreadTestB {
8. public static void main(String[] args) {
9. ExecutorService e=Executors.newFixedThreadPool(10);
10. Future f1=e.submit(new MyCallableA());
11. Future f2=e.submit(new MyCallableA());
12. Future f3=e.submit(new MyCallableA());
13. System.out.println("--Future.get()....");
14. try {
15. System.out.println(f1.get());
16. System.out.println(f2.get());
17. System.out.println(f3.get());
18. } catch (InterruptedException e1) {
19. e1.printStackTrace();
20. } catch (ExecutionException e1) {
21. e1.printStackTrace();
22. }
23. e.shutdown();
24. }
25. }
26. class MyCallableA implements Callable<String>{
27. public String call() throws Exception {
28. System.out.println("开始执行Callable");
29. String[] ss={"zhangsan","lisi"};
30. long[] num=new long[2];
31. for(int i=0;i<1000000;i++){
32. num[(int)(Math.random()*2)]++;
33. }
34. if(num[0]>num[1]){
35. return ss[0];
36. }else if(num[0]<num[1]){
37. throw new Exception("弃权!");
38. }else{
39. return ss[1];
40. }
41. }
42. }

来源：http://junlas.iteye.com/blog/846457

Java代码  

1. /**
2. * Factory and utility methods for {@link Executor}, {@link
3. * ExecutorService}, {@link ScheduledExecutorService}, {@link
4. * ThreadFactory}, and {@link Callable} classes defined in this
5. * package. This class supports the following kinds of methods:
6. *
7. * <ul>
8. *   <li> Methods that create and return an {@link ExecutorService}
9. *        set up with commonly useful configuration settings.
10. *   <li> Methods that create and return a {@link ScheduledExecutorService}
11. *        set up with commonly useful configuration settings.
12. *   <li> Methods that create and return a "wrapped" ExecutorService, that
13. *        disables reconfiguration by making implementation-specific methods
14. *        inaccessible.
15. *   <li> Methods that create and return a {@link ThreadFactory}
16. *        that sets newly created threads to a known state.
17. *   <li> Methods that create and return a {@link Callable}
18. *        out of other closure-like forms, so they can be used
19. *        in execution methods requiring <tt>Callable</tt>.
20. * </ul>
21. *
22. * @since 1.5
23. * @author Doug Lea
24. */
25. public class Executors {
26. /**
27. * Creates a thread pool that reuses a fixed number of threads
28. * operating off a shared unbounded queue.  At any point, at most
29. * <tt>nThreads</tt> threads will be active processing tasks.
30. * If additional tasks are submitted when all threads are active,
31. * they will wait in the queue until a thread is available.
32. * If any thread terminates due to a failure during execution
33. * prior to shutdown, a new one will take its place if needed to
34. * execute subsequent tasks.  The threads in the pool will exist
35. * until it is explicitly {@link ExecutorService#shutdown shutdown}.
36. *
37. * @param nThreads the number of threads in the pool
38. * @return the newly created thread pool
39. * @throws IllegalArgumentException if <tt>nThreads &lt;= 0</tt>
40. */
41. public static ExecutorService newFixedThreadPool(int nThreads) {
42. return new ThreadPoolExecutor(nThreads, nThreads,
43. 0L, TimeUnit.MILLISECONDS,
44. new LinkedBlockingQueue<Runnable>());
45. }
46. /**
47. * Creates a thread pool that reuses a fixed number of threads
48. * operating off a shared unbounded queue, using the provided
49. * ThreadFactory to create new threads when needed.  At any point,
50. * at most <tt>nThreads</tt> threads will be active processing
51. * tasks.  If additional tasks are submitted when all threads are
52. * active, they will wait in the queue until a thread is
53. * available.  If any thread terminates due to a failure during
54. * execution prior to shutdown, a new one will take its place if
55. * needed to execute subsequent tasks.  The threads in the pool will
56. * exist until it is explicitly {@link ExecutorService#shutdown
57. * shutdown}.
58. *
59. * @param nThreads the number of threads in the pool
60. * @param threadFactory the factory to use when creating new threads
61. * @return the newly created thread pool
62. * @throws NullPointerException if threadFactory is null
63. * @throws IllegalArgumentException if <tt>nThreads &lt;= 0</tt>
64. */
65. public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
66. return new ThreadPoolExecutor(nThreads, nThreads,
67. 0L, TimeUnit.MILLISECONDS,
68. new LinkedBlockingQueue<Runnable>(),
69. threadFactory);
70. }
71. /**
72. * Creates an Executor that uses a single worker thread operating
73. * off an unbounded queue. (Note however that if this single
74. * thread terminates due to a failure during execution prior to
75. * shutdown, a new one will take its place if needed to execute
76. * subsequent tasks.)  Tasks are guaranteed to execute
77. * sequentially, and no more than one task will be active at any
78. * given time. Unlike the otherwise equivalent
79. * <tt>newFixedThreadPool(1)</tt> the returned executor is
80. * guaranteed not to be reconfigurable to use additional threads.
81. *
82. * @return the newly created single-threaded Executor
83. */
84. public static ExecutorService newSingleThreadExecutor() {
85. return new FinalizableDelegatedExecutorService
86. (new ThreadPoolExecutor(1, 1,
87. 0L, TimeUnit.MILLISECONDS,
88. new LinkedBlockingQueue<Runnable>()));
89. }
90. /**
91. * Creates an Executor that uses a single worker thread operating
92. * off an unbounded queue, and uses the provided ThreadFactory to
93. * create a new thread when needed. Unlike the otherwise
94. * equivalent <tt>newFixedThreadPool(1, threadFactory)</tt> the
95. * returned executor is guaranteed not to be reconfigurable to use
96. * additional threads.
97. *
98. * @param threadFactory the factory to use when creating new
99. * threads
100. *
101. * @return the newly created single-threaded Executor
102. * @throws NullPointerException if threadFactory is null
103. */
104. public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) {
105. return new FinalizableDelegatedExecutorService
106. (new ThreadPoolExecutor(1, 1,
107. 0L, TimeUnit.MILLISECONDS,
108. new LinkedBlockingQueue<Runnable>(),
109. threadFactory));
110. }
111. /**
112. * Creates a thread pool that creates new threads as needed, but
113. * will reuse previously constructed threads when they are
114. * available.  These pools will typically improve the performance
115. * of programs that execute many short-lived asynchronous tasks.
116. * Calls to <tt>execute</tt> will reuse previously constructed
117. * threads if available. If no existing thread is available, a new
118. * thread will be created and added to the pool. Threads that have
119. * not been used for sixty seconds are terminated and removed from
120. * the cache. Thus, a pool that remains idle for long enough will
121. * not consume any resources. Note that pools with similar
122. * properties but different details (for example, timeout parameters)
123. * may be created using {@link ThreadPoolExecutor} constructors.
124. *
125. * @return the newly created thread pool
126. */
127. public static ExecutorService newCachedThreadPool() {
128. return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
129. 60L, TimeUnit.SECONDS,
130. new SynchronousQueue<Runnable>());
131. }
132. /**
133. * Creates a thread pool that creates new threads as needed, but
134. * will reuse previously constructed threads when they are
135. * available, and uses the provided
136. * ThreadFactory to create new threads when needed.
137. * @param threadFactory the factory to use when creating new threads
138. * @return the newly created thread pool
139. * @throws NullPointerException if threadFactory is null
140. */
141. public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
142. return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
143. 60L, TimeUnit.SECONDS,
144. new SynchronousQueue<Runnable>(),
145. threadFactory);
146. }
147. /**
148. * Creates a single-threaded executor that can schedule commands
149. * to run after a given delay, or to execute periodically.
150. * (Note however that if this single
151. * thread terminates due to a failure during execution prior to
152. * shutdown, a new one will take its place if needed to execute
153. * subsequent tasks.)  Tasks are guaranteed to execute
154. * sequentially, and no more than one task will be active at any
155. * given time. Unlike the otherwise equivalent
156. * <tt>newScheduledThreadPool(1)</tt> the returned executor is
157. * guaranteed not to be reconfigurable to use additional threads.
158. * @return the newly created scheduled executor
159. */
160. public static ScheduledExecutorService newSingleThreadScheduledExecutor() {
161. return new DelegatedScheduledExecutorService
162. (new ScheduledThreadPoolExecutor(1));
163. }
164. /**
165. * Creates a single-threaded executor that can schedule commands
166. * to run after a given delay, or to execute periodically.  (Note
167. * however that if this single thread terminates due to a failure
168. * during execution prior to shutdown, a new one will take its
169. * place if needed to execute subsequent tasks.)  Tasks are
170. * guaranteed to execute sequentially, and no more than one task
171. * will be active at any given time. Unlike the otherwise
172. * equivalent <tt>newScheduledThreadPool(1, threadFactory)</tt>
173. * the returned executor is guaranteed not to be reconfigurable to
174. * use additional threads.
175. * @param threadFactory the factory to use when creating new
176. * threads
177. * @return a newly created scheduled executor
178. * @throws NullPointerException if threadFactory is null
179. */
180. public static ScheduledExecutorService newSingleThreadScheduledExecutor(ThreadFactory threadFactory) {
181. return new DelegatedScheduledExecutorService
182. (new ScheduledThreadPoolExecutor(1, threadFactory));
183. }
184. /**
185. * Creates a thread pool that can schedule commands to run after a
186. * given delay, or to execute periodically.
187. * @param corePoolSize the number of threads to keep in the pool,
188. * even if they are idle.
189. * @return a newly created scheduled thread pool
190. * @throws IllegalArgumentException if <tt>corePoolSize &lt; 0</tt>
191. */
192. public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
193. return new ScheduledThreadPoolExecutor(corePoolSize);
194. }
195. /**
196. * Creates a thread pool that can schedule commands to run after a
197. * given delay, or to execute periodically.
198. * @param corePoolSize the number of threads to keep in the pool,
199. * even if they are idle.
200. * @param threadFactory the factory to use when the executor
201. * creates a new thread.
202. * @return a newly created scheduled thread pool
203. * @throws IllegalArgumentException if <tt>corePoolSize &lt; 0</tt>
204. * @throws NullPointerException if threadFactory is null
205. */
206. public static ScheduledExecutorService newScheduledThreadPool(
207. int corePoolSize, ThreadFactory threadFactory) {
208. return new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);
209. }
210. ........
211. /** Cannot instantiate. */
212. private Executors() {}
213. }