Java并发编程之Callable、Future和FutureTask的代码演示 *

概念

​

获取子线程返回的结果

UseCallable

import java.util.concurrent.Callable;

public class UseCallable implements Callable<Integer> {

private int sum;

@Override
public Integer call() throws Exception {
System.out.println("Callable子线程开始计算了");
for (int i = 0; i < 5000; i++) {
sum = sum + i;

}
System.out.println("Callable子线程计算结束!结果为 : " + sum);

return sum;
}
}

Demo01

import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;

public class Demo01 {

public static void main(String[] args) {
FutureTask<Integer> integerFutureTask = new FutureTask<>(new UseCallable());

new Thread(integerFutureTask).start();
try {
/*当FutureTask处于未启动或已启动状态时，执行FutureTask.get()方法将导致调用线程阻塞；
当FutureTask处于已完成状态时，执行FutureTask.get()方法将导致调用线程立即返回结果或抛出异常。
*/
Integer integer = integerFutureTask.get();
System.out.println("任务是否完成 : " + integerFutureTask.isDone());
System.out.println("子线程返回过来的结果是 = " + integer);


} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}

}
}

控制台打印:

Callable子线程开始计算了
Callable子线程计算结束!结果为 : 12497500
任务是否完成 : true
子线程返回过来的结果是 = 12497500

获取子线程返回的实体类

Entity

public class Entity {
private String name;
private int sum;

//**********************************


@Override
public String toString() {
return "Entity{" +
"name='" + name + '\'' +
", sum=" + sum +
'}';
}

public String getName() {
return name;
}

public void setName(String name) {
this.name = name;
}

public int getSum() {
return sum;
}

public void setSum(int sum) {
this.sum = sum;
}
}

TestEntityCallable

import entity.Entity;

import java.util.concurrent.Callable;

public class TestEntityCallable implements Callable<Entity> {
@Override
public Entity call() {
Entity entity = new Entity();
entity.setName("这个是实体的名字");
entity.setSum(1111111);
return entity;
}
}

Demo02

import entity.Entity;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;

public class Demo02 {
public static void main(String[] args) {

TestEntityCallable testEntityCallable = new TestEntityCallable();

FutureTask<Entity> entityFutureTask = new FutureTask<>(testEntityCallable);
new Thread(entityFutureTask).start();
try {
Entity entity = entityFutureTask.get();

System.out.println("子线程返回的结果是 = " + entity);
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}

}
}

控制台打印结果:

子线程返回的结果是 = Entity{name='这个是实体的名字', sum=1111111}

终止任务

TaskCallable

public class TaskCallable implements Callable<String> {

@Override
public String call() throws Exception {
while (true) {
System.out.println("子线程一直在执行\n");
Thread.sleep(100);

if (Thread.currentThread().isInterrupted()) {
//在while 死循环 没有执行到这里
System.out.println("主线程终止了我");
return null;

}

}

}

}

Demo01

import utils.SleepTools;

import java.util.concurrent.*;

/**
* 使用Future的cancel()方法来取消已经提交给执行者的任务
*/
public class Demo01 {

public static void main(String[] args) {
/*Task实现了Callable接口*/
FutureTask futureTask = new FutureTask<>(new TaskCallable());
new Thread(futureTask).start();
/*
* 那么cancel是如何工作的呢？
*
* 当你想要取消你已提交给执行者的任务，使用Future接口的cancel()方法。
* 根据cancel()方法参数和任务的状态不同，这个方法的行为将不同：
* 1、如果这个任务已经完成或之前的已经被取消或由于其他原因不能被取消，
* 那么这个方法将会返回false并且这个任务不会被取消。
* 2、如果这个任务正在等待执行者获取执行它的线程，那么这个任务将被取消而且不会开始他的执行。
* 如果这个任务已经正在运行，则视方法的参数情况而定。
* cancel()方法接收一个Boolean值参数。
* 如果参数为true并且任务正在运行，那么这个任务将被取消。
* 如果参数为false并且任务正在运行，那么这个任务将不会被取消。
*/
SleepTools.second(3);//睡眠三秒
boolean cancel = futureTask.cancel(true);//终止任务

boolean cancelled = futureTask.isCancelled();
System.out.printf("Main取消:%s\n", cancelled);
boolean done = futureTask.isDone();

System.out.printf("任务是否完成了:%s\n", done);


}
}

控制台输出:

子线程一直在执行

子线程一直在执行

子线程一直在执行

子线程一直在执行

子线程一直在执行

Main取消:true
任务是否完成了:true

Process finished with exit code 0

一个线程获取另外一个线程的结果

RecommendTask 推荐接口

import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;

/**
* @类说明: 调用推荐接口获取数据
* 张俊杰 2020年10月21日 20:53
*/
public class RecommendTask implements Callable<String> {
@Override
public String call() throws Exception {
System.out.println("RecommendTask: 调用推荐接口获取数据...");
TimeUnit.SECONDS.sleep(1);

System.out.println("RecommendTask: 得到推荐接口数据...");
TimeUnit.SECONDS.sleep(10);
return " [RecommendTask 板块数据] ";
}
}

SearchTask 搜索Task

import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;

/**
* @类说明: SearchTask 调用搜索接口数据，同时需要t1推荐接口数据
* 张俊杰 2020年10月21日 20:54
*/
public class SearchTask implements Callable<String> {
FutureTask<String> ft1;

// SearchTask 任务需要 RecommendTask 任务的 FutureTask 返回结果去重
SearchTask(FutureTask<String> ft1) {
this.ft1 = ft1;
}

@Override
public String call() throws Exception {
System.out.println("SearchTask: 调用搜索接口获取数据...");
TimeUnit.SECONDS.sleep(1);

System.out.println("SearchTask: 得到搜索接口的数据...");
TimeUnit.SECONDS.sleep(5);
// 获取 T2 线程的数据
System.out.println("SearchTask: 调用 RecommendTask.get() 接口获取推荐数据");
String tf1 = ft1.get();
System.out.println("SearchTask: 获取到推荐接口数据:" + tf1);

System.out.println("SearchTask: 将 RecommendTask 与 SearchTask 板块数据做去重处理");
return "[RecommendTask 和 SearchTask 板块数据聚合结果]";
}
}

Demo1

import java.util.concurrent.FutureTask;

public class Demo01 {
public static void main(String[] args) throws Exception {
FutureTask<String> stringFutureTask = new FutureTask<>(new RecommendTask());
Thread T1 = new Thread(stringFutureTask);
T1.start();


FutureTask<String> ft2 = new FutureTask<>(new SearchTask(stringFutureTask));
Thread T2 = new Thread(ft2);
T2.start();
System.out.println("主线程获取结果"+ft2.get());
}


}

搜索Task需要获取推荐Task的结果的值拼接在一起, 最后主线程获取搜索Task的接口

控制台打印:

RecommendTask: 调用推荐接口获取数据...
SearchTask: 调用搜索接口获取数据...
RecommendTask: 得到推荐接口数据...
SearchTask: 得到搜索接口的数据...
SearchTask: 调用 RecommendTask.get() 接口获取推荐数据
SearchTask: 获取到推荐接口数据: [RecommendTask 板块数据]
SearchTask: 将 RecommendTask 与 SearchTask 板块数据做去重处理
主线程获取结果[RecommendTask 和 SearchTask 板块数据聚合结果]