Callable实现JAVA多线程

最近项目用到一个功能须要实现多线程分发任务且须要任务的返回值，以前一直都是实现Runnable接口，但里面的run方法是返回void的。后来在网上查了下JAVA1.5开始就有了Callable。

下面来看看如何倒腾下这个东西。

import java.util.concurrent.Callable;

/**
 * @类说明 线程业务处理
 * @author DavenTsang
 * @date 2016-11-16
 * 
 */
public class PoolTask implements Callable<String> {

    private String id;

    @Override
    public String call() throws Exception {
        return "当前线程名:" + Thread.currentThread().getName() + ":" + id;
    }

    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }

}

View Code

先创建一个类实现Callable接口。一下是JDK的API对Callable接口的描述：

public interface Callable<V>

返回结果而且可能抛出异常的任务。实现者定义了一个不带任何参数的叫作 call 的方法。

Callable 接口相似于 Runnable，二者都是为那些其实例可能被另外一个线程执行的类设计的。可是 Runnable 不会返回结果，而且没法抛出通过检查的异常。

Executors 类包含一些从其余普通形式转换成 Callable 类的实用方法。

V是须要返回的对象。

须要执行这个实现类，咱们须要建立一个线程池

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class  PoolUtils {

        //可伸缩线程池
        private static ExecutorService cachedPool = Executors.newCachedThreadPool();
        public static CompletionService<String> completionService = new ExecutorCompletionService<String>(cachedPool);
        private PoolTask task;
        public String addTask() throws InterruptedException, ExecutionException{
            //添加任务
            Future<String> future = completionService.submit(task);
            //检查是否出现第二个线程进来
            Thread.sleep(1000);
            List<Future<String>> list = new ArrayList<Future<String>>();
            System.out.println(completionService.take().get());
            list.add(future);
            return completionService.take().get();
        }
        
        public PoolTask getTask() {
            return task;
        }
        public void setTask(PoolTask task) {
            this.task = task;
        }
        
}

View Code

Thread.sleep();调用这个方法是由于在前面添加任务是用一个线程数组调用，看下Executors.newCachedThreadPool();这个是否能够本身去根据须要建立线程。

咱们再来看下submit()方法的源码，

public Future<V> submit(Callable<V> task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<V> f = newTaskFor(task);
        executor.execute(new QueueingFuture(f));
        return f;
    }

 

  private class QueueingFuture extends FutureTask<Void> {
        QueueingFuture(RunnableFuture<V> task) {
            super(task, null);
            this.task = task;
        }
        protected void done() { completionQueue.add(task); }
        private final Future<V> task;
    }

 追一下源码会发现里面有个队列存在。submit添加一个任务就会放到队列里面。这样就不用咱们显示的去建立一个队列。

调用类：

import java.util.concurrent.ExecutionException;

public class Test {
    
    
    public static void main(String[] args) throws InterruptedException, ExecutionException {
            Thread[] thread = new Thread[10];
            for(int i = 0;i<thread.length;i++){
                thread[i] = new Thread(new A());
            }
            for(Thread th : thread){
                th.start();
            }
        
        
    }
}

class A implements Runnable{
    @Override
    public void run() {
        PoolTask task = new PoolTask();
        task.setId("daven");
        PoolUtils utils = new PoolUtils();
        utils.setTask(task);
        try {
            String a = utils.addTask();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }
}

View Code

调用类用多个线程去调用是由于模拟项目的场景了。

以上是一些记录供本身回忆用。遇到问题先本身查找缘由，再去网上找下而后再找API追源码。