Java并发：Callable与Future的应用

系统管理员 2022-06-10 04:19 201阅读 0赞

我们都知道实现多线程有3种方式，一种是继承Thread，一种是实现Runnable，但这2种方式都有一个缺陷，在任务完成后无法获取返回结果，实际使用的很少。要想获得返回结果，就得使用Callable，Callable任务可以有返回值，通常的用法是需要多个callable异步去执行，将Callable提交到ThreadPoolExecutor 中后，我们就能得到Future对象，后面根据Future去获取Callable的返回结果。

Runable只有一个返回值为void的run方法，且无法抛出异常，适用于不需要返回值的场景

public interface Runnable {
        /**
         * When an object implementing interface <code>Runnable</code> is used
         * to create a thread, starting the thread causes the object's
         * <code>run</code> method to be called in that separately executing
         * thread.
         * <p>
         * The general contract of the method <code>run</code> is that it may
         * take any action whatsoever.
         *
         * @see     java.lang.Thread#run()
         */
        public abstract void run();
    }

Callable中有一个call()函数，同时有一个泛型的返回值，而且可以抛出异常，适用于需要返回值的场景。

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

Future就是对于具体的Runnable或者Callable任务的执行结果进行取消、查询是否完成、获取结果、设置结果操作。get方法会阻塞，直到任务返回结果

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

示例

报表生成接口，往往需要从n多张表里统计数据，假设一个报表需要从4张表来统计数据，1个线程顺序去执行4个查询，1个查询按200ms，4个差不多就800ms，但若是改成4个线程并发去执行，耗时就能大大缩短。这个时候就要用Callabel结合Future来实现。以下代码给出了使用的示例。

ThreadPoolExecutor threadPool = new ThreadPoolExecutor(4, 4, 0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>());
    //查询合伙人订单信息
    long startTime=System.currentTimeMillis();
    Callable<List<SetOfAccountsSelectPlatRes>> businessCallable= () -> platformAccountMapper.setOfAccountsSelectPartnerBus(param);
    Future businessFuture = threadPool.submit(businessCallable);
    //查询合伙人信息
    Callable<List<SetOfAccountsSelectPlatRes>> partnerInfoCallable= () -> platformAccountMapper.setlectPartnerInfo(param);
    Future partnerInfoFuture = threadPool.submit(partnerInfoCallable);
    //查询合伙人提现服务费
    Callable<List<SetOfAccountsSelectPlatRes>> feeAmountCallable= () -> platformAccountMapper.setOfAccountsSelectPartnerFeeAmount(param);
    Future feeAmountFuture = threadPool.submit(feeAmountCallable);
    //查询合伙人已提现信息
    Callable<List<SetOfAccountsSelectPlatRes>> cashCallable= () -> platformAccountMapper.setOfAccountsSelectPartnerTotalCash(param);
    Future cashFuture = threadPool.submit(cashCallable);
    threadPool.shutdown();
    
    List<SetOfAccountsSelectPlatRes> setOfAccountsSelectPartnerResBusList = ( List<SetOfAccountsSelectPlatRes>)businessFuture.get();
    long partnerBusinessoQueryEndTime=System.currentTimeMillis();
    if (CollectionUtils.isNotEmpty(setOfAccountsSelectPartnerResBusList)){
        log.info("合伙人套账导出查询到合伙人订单信息耗时: {}", (partnerBusinessoQueryEndTime-startTime));
    }
    List<SetOfAccountsSelectPlatRes> partnerInfoList = ( List<SetOfAccountsSelectPlatRes>)partnerInfoFuture.get();
      List<SetOfAccountsSelectPlatRes> setOfAccountsSelectPartnerResFeeAmountList = (List<SetOfAccountsSelectPlatRes>)feeAmountFuture.get();
    long partnerResFeeQueryEndTime=System.currentTimeMillis();
    if (CollectionUtils.isNotEmpty(setOfAccountsSelectPartnerResFeeAmountList)){
       log.info("合伙人套账导出查询到合伙人提现服务费信息耗时: {}", (partnerResFeeQueryEndTime-startTime));
    }
    List<SetOfAccountsSelectPlatRes> setOfAccountsSelectPartnerResTotalCashtList = (List<SetOfAccountsSelectPlatRes>)cashFuture.get();
     long partnerCashQueryEndTime=System.currentTimeMillis();
    if (CollectionUtils.isNotEmpty(setOfAccountsSelectPartnerResTotalCashtList)){
        log.info("合伙人套账导出查询到合伙人已提现信息耗时: {}", (partnerCashQueryEndTime-startTime));
    }