重拾JAVA线程之获取另一个线程的返回

发表于 更新于 分类于 ， ，

前言

Java的多线程机制允许我们将可以并行的任务分配给不同的线程同时完成。但是，如果我们希望在另一个线程的结果之上进行后续操作，我们应该怎么办呢？

注：本文的代码没有经过具体实践的检验，纯属为了展示。如果有任何问题，欢迎指出。

在此之前你需要了解

  • Thread类
  • Runnable接口
  • ExecutorServer, Executors生成的线程池

一个简单的场景

假设我们现在有一个IO操作需要读取一个文件，在读取完成之后我们希望针对读取的字节进行相应的处理。因为IO操作比较耗时，所以我们可能会希望在另一个线程中进行IO操作，从而确保主线程的运行不会出现等待。在这里，我们读取完文件之后会在其所在线程输出其字符流对应的字符串。

1
2
3
4
5
6
7
8
9
10
11
12
13
//主线程类
public class MainThread {

    public static void main(String[] args){
        performIO();
    }

    public static void performIO(){
        FileReader fileReader = new FileReader(FILENAME);
        Thread thread = new Thread(fileReader);
        thread.start();
    }
}

文件读取类：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
public class FileReader implements  Runnable{

    private FileInputStream fileInputStream;
    private String fileName;

    private byte[] content;

    public FileReader(String fileName){
        this.fileName = fileName;
        content = new byte[2048];
    }

    @Override
    public void run() {
        try {
            File file = new File(fileName);
            fileInputStream = new FileInputStream(file);
            int bytesRead = 0;
            while(fileInputStream.available() > 0){
                bytesRead += fileInputStream.read(content, bytesRead, content.length - bytesRead);
            }
            System.out.println(new String(content,0, bytesRead));
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

一个错误的例子

假设现在主线程希望针对文件的信息进行操作，那么可能会出现以下的代码：

在子线程中添加get方法返回读取的字符数组：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
public class FileReader implements Runnable{

    private FileInputStream fileInputStream;
    private String fileName;

    private byte[] content;

    //添加get方法返回字符数组
    public byte[] getContent(){
        return this.content;
    }
    
    public FileReader(String fileName){
        this.fileName = fileName;
        content = new byte[2048];
    }

    @Override
    public void run() {
        try {
            File file = new File(fileName);
            fileInputStream = new FileInputStream(file);
            int bytesRead = 0;
            while(fileInputStream.available() > 0){
                bytesRead += fileInputStream.read(content, bytesRead, content.length - bytesRead);
            }
            System.out.println(new String(content,0, bytesRead));
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

主线程方法中添加读取byte数组的方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
public class MainThread {

    public static void main(String[] args){
        performIO();
    }

    public static void performIO(){
        FileReader fileReader = new FileReader(FILENAME);
        Thread thread = new Thread(fileReader);
        thread.start();

        //读取内容
        byte[] content = fileReader.getContent();
        System.out.println(content);
    }
}

这段代码不能保证正常运行，原因在于我们无法控制线程的调度。也就是说，在thread.start()语句后，主线程可能依然占有CPU继续执行，而此时获得的content则是null。

你搞定了没有啊

主线程可以通过轮询的方式询问IO线程是不是已经完成了操作，如果完成了操作，就读取结果。这里我们需要设置一个标记位来记录IO是否完成。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
public class FileReader implements Runnable{

    private FileInputStream fileInputStream;
    private String fileName;

    private byte[] content;
    //新建标记位，初始为false
    public boolean finish;
    
    public byte[] getContent(){
        return this. content;
    }
    public FileReader(String fileName){
        this.fileName = fileName;
        content = new byte[2048];
    }

    @Override
    public void run() {
        try {
            File file = new File(fileName);
            fileInputStream = new FileInputStream(file);
            int bytesRead = 0;
            while(fileInputStream.available() > 0){
                bytesRead += fileInputStream.read(content, bytesRead, content.length - bytesRead);
            }
            finish = true;
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

主线程一直轮询IO线程：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
public class MainThread {

    public static void main(String[] args){
        performIO();
    }
    
    public static void performIO(){
        FileReader fileReader = new FileReader(FILENAME);
        Thread thread = new Thread(fileReader);
        thread.start();
        while(true){
            if(fileReader.finish){
                System.out.println(new String(fileReader.getContent()));
                break;
            }
        }
    }
}

缺点那是相当的明显，不断的轮询会无谓的消耗CPU。除此以外，一旦IO异常，则标记位永远为false，主线程会陷入死循环。

搞定了告诉我一声啊

要解决这个问题，我们就需要在IO线程完成读取之后，通知主线程该操作已经完成，从而主线程继续运行。这种方法叫做回调。可以用静态方法实现：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
public class FileReader implements Runnable{

    private FileInputStream fileInputStream;
    private String fileName;

    private byte[] content;
    
    public FileReader(String fileName){
        this.fileName = fileName;
        content = new byte[2048];
    }

    @Override
    public void run() {
        try {
            File file = new File(fileName);
            fileInputStream = new FileInputStream(file);
            int bytesRead = 0;
            while(fileInputStream.available() > 0){
                bytesRead += fileInputStream.read(content, bytesRead, content.length - bytesRead);
            }
            
            //完成IO后调用主线程的回调函数来通知主线程进行后续的操作
            MainThread.callback(content);
            
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

主线程方法中定义回调函数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
public class MainThread {

    public static void main(String[] args){
        performIO();
    }

    //在主线程中用静态方法定义回调函数
    public static void callback(byte[] content){
        //do something
        System.out.println(content);
    }
    
    public static void performIO(){
        FileReader fileReader = new FileReader(FILENAME);
        Thread thread = new Thread(fileReader);
        thread.start();
    }
}

这种实现方法的缺点在于MainThread和FileReader类之间的耦合太强了。而且万一我们需要读取多个文件，我们会希望对每一个FileReader有自己的callback函数进行处理。因此我们可以callback将其声明为一般函数，并且让IO线程持有需要回调的方法所在的实例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
public class FileReader implements Runnable{

    private FileInputStream fileInputStream;
    private String fileName;

    private byte[] content;

    //持有回调函数的实例
    private MainThread mainThread;
   
    //传入实例
    public FileReader(String fileName, MainThreand mainThread){
        this.fileName = fileName;
        content = new byte[2048];
        this.mainThread = mainThread;
    }

    @Override
    public void run() {
        try {
            File file = new File(fileName);
            fileInputStream = new FileInputStream(file);
            int bytesRead = 0;
            while(fileInputStream.available() > 0){
                bytesRead += fileInputStream.read(content, bytesRead, content.length - bytesRead);
            }
            System.out.println(new String(content,0, bytesRead));
            mainThread.callback(content);
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

主线程方法中添加读取byte数组的方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
public class MainThread {

    public static void main(String[] args){
        new MainThread().performIO();
    }

    public void callback(byte[] content){
        //do something
    }
    
    //将执行IO变为非静态方法
    public void performIO(){
        FileReader fileReader = new FileReader(FILENAME);
        Thread thread = new Thread(fileReader);
        thread.start();
    }
}

搞定了告诉我们一声啊

有时候可能有多个事件都在监听事件，比如当我点击了Button，我希望后台能够执行查询操作并将结果返回给UI。同时,我还希望将用户的这个操作无论成功与否写入日志线程。因此，我可以写两个回调函数，分别对应于不同的操作。

1
2
3
public interface Callback<T>{
    public void perform(T t);
}

写入日志操作：

1
2
3
4
5
public class Log implements Callback<String>{
    public void perform(String s){
        //写入日志
    }
}

IO读取操作

1
2
3
4
5
public class FileReader implements Callback<String>{
    public void perform(String s){
        //进行IO操作
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
public class Button{
    private List<Callable> callables;
    
    public Button(){
        callables = new ArrayList<Callable>();
    }
    
    public void addCallable(Callable c){
        this.callables.add(c);
    }
    
    public void onClick(){
        for(Callable c : callables){
            c.perform(...);
        }
    }
}

Java7: 行了，别忙活了，朕知道了

Java7提供了非常方便的封装Future，Callables和Executors来实现之前的回调工作。

之前我们直接将任务交给一个新建的线程来处理。可是如果每次都新建一个线程来处理当前的任务，线程的新建和销毁将会是一大笔开销。因此Java提供了多种类型的线程池来供我们操作。它将管理线程的创建销毁和复用，尽最大可能提高线程的使用效率。

同时Java7提供的Callable接口将自动返回线程运行结束的结果。如果我们在另一个线程中需要使用这个结果，则这个线程会挂起直到另一个线程返回该结果。我们无需再在另一个线程中使用回调函数来处理结果。

假设现在我们想要找到一个数组的最大值。假设该数组容量惊人，因此我们希望新开两个线程分别对数组的前半部分和后半部分计算最大值。然后在主线程中比较两个结果得出结论：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
public class ArrayMaxValue {

    public static void main(String[] args){

        Random r = new Random(20);
        int[] array = new int[500];
        for (int i = 0 ; i<array.length ; i++){
            array[i] = r.nextInt();
        }
        ArrayMaxValue a = new ArrayMaxValue();
        a.max(array);
    }

    public void max(int[] array){
        ExecutorService executorService = Executors.newCachedThreadPool();
        int mid = array.length / 2;
        Future<Integer> f1 = executorService.submit(new MaxValue(array, 0, mid));
        Future<Integer> f2 = executorService.submit(new MaxValue(array, mid, array.length));

        try {
            //主线程将阻塞自己直到两个线程都完成运行，并返回结果
            System.out.println(Math.max(f1.get(), f2.get()));
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }
    public class MaxValue implements Callable<Integer>{
        private final int[] array;
        private final int startIndex;
        private final int endIndex;
        public MaxValue(int[] array, int startIndex, int endIndex){
            this.array = array;
            this.startIndex = startIndex;
            this.endIndex = endIndex;
        }
        @Override
        public Integer call() throws Exception {
            int max = Integer.MIN_VALUE;
            for (int i = startIndex ; i<endIndex ; i++){
                max = Math.max(max, array[i]);
            }
            System.out.println(max);
            return max;
        }
    }
}

参考文章

深入理解线程通信