Java并发编程之Callable与Future

﹏ヽ暗。殇╰゛Y 2022-06-02 05:19 197阅读 0赞

***java.util.concurrent.Callable***是JDK1.5引入的线程接口，与Runnable类似，但可以返回一个对象，也可以抛出异常。

@FunctionalInterface
    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;
    }

Callable使用泛型，因此它可以返回任意类型的对象。**Executor**框架提供了submit()方法在线程池中调用执行Callable实现，Executor提供的是工作线程模式，你可以通过

> **Executors.newFixedThreadPool(10)**

来初始化线程池中的工作线程，之后可以提交任务来调用线程处理。如果没有空闲线程，则任务需要等待执行。作为线程执行目标，“任务”需要实现**Runnable**接口或**Callable**接口，两者不同的是，Callable类型任务可以返回任务执行结果或执行状态。

***java.util.concurrent.Future*** 定义如下：

public interface Future<V> {
    
        /**
         * Attempts to cancel execution of this task.  This attempt will
         * fail if the task has already completed, has already been cancelled,
         * or could not be cancelled for some other reason. If successful,
         * and this task has not started when {@code cancel} is called,
         * this task should never run.  If the task has already started,
         * then the {@code mayInterruptIfRunning} parameter determines
         * whether the thread executing this task should be interrupted in
         * an attempt to stop the task.
         *
         * <p>After this method returns, subsequent calls to {@link #isDone} will
         * always return {@code true}.  Subsequent calls to {@link #isCancelled}
         * will always return {@code true} if this method returned {@code true}.
         *
         * @param mayInterruptIfRunning {@code true} if the thread executing this
         * task should be interrupted; otherwise, in-progress tasks are allowed
         * to complete
         * @return {@code false} if the task could not be cancelled,
         * typically because it has already completed normally;
         * {@code true} otherwise
         */
        boolean cancel(boolean mayInterruptIfRunning);
    
        /**
         * Returns {@code true} if this task was cancelled before it completed
         * normally.
         *
         * @return {@code true} if this task was cancelled before it completed
         */
        boolean isCancelled();
    
        /**
         * Returns {@code true} if this task completed.
         *
         * Completion may be due to normal termination, an exception, or
         * cancellation -- in all of these cases, this method will return
         * {@code true}.
         *
         * @return {@code true} if this task completed
         */
        boolean isDone();
    
        /**
         * Waits if necessary for the computation to complete, and then
         * retrieves its result.
         *
         * @return the computed result
         * @throws CancellationException if the computation was cancelled
         * @throws ExecutionException if the computation threw an
         * exception
         * @throws InterruptedException if the current thread was interrupted
         * while waiting
         */
        V get() throws InterruptedException, ExecutionException;
    
        /**
         * Waits if necessary for at most the given time for the computation
         * to complete, and then retrieves its result, if available.
         *
         * @param timeout the maximum time to wait
         * @param unit the time unit of the timeout argument
         * @return the computed result
         * @throws CancellationException if the computation was cancelled
         * @throws ExecutionException if the computation threw an
         * exception
         * @throws InterruptedException if the current thread was interrupted
         * while waiting
         * @throws TimeoutException if the wait timed out
         */
        V get(long timeout, TimeUnit unit)
            throws InterruptedException, ExecutionException, TimeoutException;
    }

当我们通过Executor的submit方法向线程池提交Callable任务时，方法返回的是***java.util.concurrent.Future***对象, Future对象关联的是我们提交的Callable任务，通过Future提供方法，我们可以“实时探测”线程池中已提交的Callable任务的执行状态或对Callable任务进行干预。

>  **Future submit(Callable task);**

<table> 
 <thead> 
  <tr> 
   <th>方法</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>cancel</td> 
   <td>取消关联Callable任务</td> 
  </tr> 
  <tr> 
   <td>get</td> 
   <td>获取任务状态或结果，注意，该方法为同步方法，方法调用方在调用get方法之后，会一直等待任务执行完毕，因此为了避免调用方长时间等待，该方法包含了一个重载方法，允许调用方通过指定超时时间以避免调用方长时间阻塞。</td> 
  </tr> 
  <tr> 
   <td>isDone</td> 
   <td>Callable任务是否执行完毕</td> 
  </tr> 
  <tr> 
   <td>isCancelled</td> 
   <td>Callable任务是否已取消</td> 
  </tr> 
 </tbody> 
</table>

### **示例代码** ###

*  创建求和偏函数Callable任务

public class CallableAdder implements Callable<Integer> { 
        Integer operand1;
        Integer operand2;
    
        CallableAdder(Integer operand1, Integer operand2) {
            this.operand1 = operand1;
            this.operand2 = operand2;
        }
    
        public Integer call() throws Exception {
            Thread.sleep(1000);
            System.out.println(Thread.currentThread().getName() + " says : partial Sum for " + operand1 + " and " + operand2 + " is " + (operand1 + operand2));
            return operand1 + operand2;
        }
    }

*  创建管理类，负责对整数列表进行分组，并提交分组给Executor框架

public class ParallelAdder {
        public Integer completionParallelSum() {
            StopWatch stopWatch = new StopWatch();
            stopWatch.start();
            ExecutorService executor = Executors.newFixedThreadPool(10);
            CompletionService<Integer> pool = new ExecutorCompletionService<>(executor);
            int count = 1;
            int prev = 0;
            int num = 0;
            for (int i = 1; i < 101; i++) {
                if (count % 2 == 0) {
                    //grouping
                    //System.out.println("Prev :" + prev + " current: " + i);
                    pool.submit(new CallableAdder(prev, i));
                    num++;
                    count = 1;
                    continue;
                }
                prev = i;
                count++;
            }
            int totalSum = 0;
            System.out.println(num);
            for (int i = 0; i < num; i++) {
                try {
                    totalSum += pool.take().get();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }
            }
            System.out.println("Total Sum is " + totalSum);
            System.out.println("Time taken by completionParallelSum " + stopWatch.getTime());
            executor.shutdown();
            return totalSum;
        }
    
        public Integer parallelSum() {
            StopWatch stopWatch = new StopWatch();
            stopWatch.start();
            ExecutorService executor = Executors.newFixedThreadPool(10);
            List<Future<Integer>> list = new ArrayList<>();
            int count = 1;
            int prev = 0;
            for (int i = 1; i < 101; i++) {
                if (count % 2 == 0) {
                    //grouping
                    //System.out.println("Prev :" + prev + " current: " + i);
                    Future<Integer> future = executor.submit(new CallableAdder(prev, i));
                    list.add(future);
                    count = 1;
                    continue;
                }
                prev = i;
                count++;
            }
            int totalSum = 0;
            for (Future<Integer> future : list) {
                try {
                    totalSum = totalSum + future.get();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }
            }
            System.out.println("Total Sum is " + totalSum);
            System.out.println("Time taken by parallelSum " + stopWatch.getTime());
            executor.shutdown();
            return totalSum;
        }
    
        public int sequentialSum() {
            StopWatch stopWatch = new StopWatch();
            stopWatch.start();
            Integer totalSum = 0;
            for (int i = 1; i < 101; i++) {
                totalSum = totalSum + i;
            }
            System.out.println("sequentialSum Total Sum is " + totalSum);
            System.out.println("Time taken by sequentialSum " + stopWatch.getTime());
            return totalSum;
        }
    
        public static void main(String[] args) {
            ParallelAdder adder = new ParallelAdder();
            int pSum = adder.parallelSum();
            int sSum = adder.sequentialSum();
            int cSum = adder.completionParallelSum();
            System.out.println("parallel Sum equals to Sequential Sum ? ");
            System.out.println("Answer is :: " + (pSum == sSum));
            System.out.println("Answer is :: " + (pSum == cSum));
        }
    }

[https://dzone.com/articles/java-callable-future-understanding][https_dzone.com_articles_java-callable-future-understanding]

[https_dzone.com_articles_java-callable-future-understanding]: https://dzone.com/articles/java-callable-future-understanding