第四章 Thread Executors(Executors多线程架构)【下】

时间：2022-09-14 18:35:13
本章涉及内容：
  • 创建一个executor线程
  • 创建一个固定大小executor
  • 执行executor任务返回一个结果
  • 处理多任务和处理第一个结果
  • 处理多任务和处理所有结果
  • 延迟之后运行executor
  • 周期地运行任务的executor
  • 在executor取消任务
  • 在executor控制任务完成
  • 在executor分开task启动和处理结果
  • 控制executor的拒绝的任务

1、在executor控制任务完成之后继续做一些事情

重写FutureTask类来改写它们默认的行为

package com.jack;

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

public class ExecutableTask implements Callable<String> {

private String name;


public String getName() {
return name;
}

public ExecutableTask(String name) {
super();
this.name = name;
}



@Override
public String call() throws Exception {
try{
long duration = (long) (Math.random()*10);
System.out.printf("%s: 等待 %d 秒\n", this.name, duration);
TimeUnit.SECONDS.sleep(duration);
} catch (InterruptedException e){

}
return "Hello, world, I'm " + name;
}

}

package com.jack;

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

public class ResultTask extends FutureTask<String>{

private String name;

public ResultTask(Callable<String> callable) {
super(callable);
this.name = ((ExecutableTask)callable).getName();
}

@Override
protected void done() {
if (isCancelled()){
System.out.printf("%s： 已经取消了\n", name);
} else {
System.out.printf("%s: 已经完成了\n", name);
}
}


}

package com.jack;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class Main {
public static void main(String[] args) {
ExecutorService executor = Executors.newCachedThreadPool();
ResultTask resultTasks[] = new ResultTask[5];
for (int i=0; i<5; i++){
ExecutableTask executableTask = new ExecutableTask("Task " + i);
resultTasks[i] = new ResultTask(executableTask);
executor.submit(resultTasks[i]);
}
try{
TimeUnit.SECONDS.sleep(5);
}catch (InterruptedException e){
e.printStackTrace();
}
for (int i=0; i<resultTasks.length; i++){
resultTasks[i].cancel(true);
}
for (int i=0; i<resultTasks.length; i++){
try{
if(!resultTasks[i].isCancelled()){
System.out.printf("%s\n", resultTasks[i].get());
}
} catch(InterruptedException | ExecutionException e){
e.printStackTrace();
}
}
executor.shutdown();
}
}

日志： 

Task 0: 等待 7 秒
Task 1: 等待 2 秒
Task 3: 等待 5 秒
Task 2: 等待 0 秒
Task 4: 等待 4 秒
Task 2: 已经完成了
Task 1: 已经完成了
Task 4: 已经完成了
Task 0： 已经取消了
Task 3： 已经取消了
Hello, world, I'm Task 1
Hello, world, I'm Task 2
Hello, world, I'm Task 4


总结：

  • 1、首先创建一个执行者。ExecutorService
  • 2、创建继承FutureTask<V>的类的数组，并重写了done()方法
  • 3、将会实现Callable接口任务注入到FutureTask数组中。
  • 4、executor进行执行
  • 5、休眠5秒，如果能在5秒执行完都会执行完，其他都会取消
  • 6、执行取消命令
  • 7、对于没有取消获取返回的字符串也就是“Hello world”

2、分开处理启动任务和他们的结果

这里采用的是CompletionService类进行处理

package com.jack;

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

public class ReportGenerator implements Callable<String>{

private String sender;
private String title;

public ReportGenerator(String sender, String title) {
super();
this.sender = sender;
this.title = title;
}

@Override
public String call() throws Exception {
try{
Long duration = (long) (Math.random()*10);
System.out.printf("%s_%s: ReportGenerator：生成一个报告持续了%d 秒\n", this.sender, this.title, duration);
TimeUnit.SECONDS.sleep(duration);
} catch (InterruptedException e){
e.printStackTrace();
}
String ret = sender+": "+ title;
return ret;
}

}

package com.jack;

import java.util.concurrent.CompletionService;

public class ReportRequest implements Runnable{

private String name;

private CompletionService<String> service;

public ReportRequest(String name, CompletionService<String> service) {
super();
this.name = name;
this.service = service;
}

@Override
public void run() {
ReportGenerator reportGenerator = new ReportGenerator(name, "Report");
service.submit(reportGenerator);
}

}

package com.jack;

import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

public class ReportProcessor implements Runnable {

private CompletionService<String> service;
private boolean end;


public ReportProcessor(CompletionService<String> service) {
super();
this.service = service;
end = false;
}


@Override
public void run() {
while (!end){
try{
Future<String> result = service.poll(20, TimeUnit.SECONDS);
if(result!=null){
String report = result.get();
System.out.printf("ReportReceiver: 报告已经接收：%s\n", report);

} 
}catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
System.out.printf("ReportSender: 结束\n");
}
}


public void setEnd(boolean end) {
this.end = end;
}

}

package com.jack;

import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class Main {
public static void main(String[] args) {
ExecutorService executor = Executors.newCachedThreadPool();
CompletionService<String> service = new ExecutorCompletionService<>(executor);
ReportRequest faceRequest = new ReportRequest("Face " , service);
ReportRequest onlineRequest = new ReportRequest("Online", service);
Thread faceThread = new Thread(faceRequest);
Thread onlineThread = new Thread(onlineRequest);
ReportProcessor processor = new ReportProcessor(service);
Thread senderThread = new Thread(processor);
System.out.printf("Main: 开启线程\n");
faceThread.start();
onlineThread.start();
senderThread.start();
try{
System.out.printf("Main: 等待报道生成\n");
faceThread.join();
onlineThread.join();
} catch (InterruptedException e){
e.printStackTrace();
}

System.out.printf("Main : 关闭executor\n");
executor.shutdown();
try{
executor.awaitTermination(1, TimeUnit.DAYS);
} catch (InterruptedException e){
e.printStackTrace();
}
processor.setEnd(true);
System.out.printf("Main: 结束了");
}
}

日志： 

Main: 开启线程
Main: 等待报道生成
Main : 关闭executor
Face _Report: ReportGenerator：生成一个报告持续了7 秒
Online_Report: ReportGenerator：生成一个报告持续了9 秒
ReportReceiver: 报告已经接收：Face : Report
ReportSender: 结束
ReportReceiver: 报告已经接收：Online: Report
ReportSender: 结束
Main: 结束了
总结：

  • 1、创建一个执行线程ExecutorService executor
  • 2、创建一个CompletionService<String> 包装executor的类
  • 3、创建两个请求的线程，service.submit() 来说生成报告
  • 4、创建一个线程发送线程，其中里面有个主要的方法Future<String> result = service.poll(20, TimeUnit.SECONDS); 
  • 这句话的意思就是：两个请求产生结果将会放置到一个队列，而发送线程在该队列中取用，参数意思是等待时间20秒，如果20秒没有结果不在等待继续下面执行
  • 5、faceThread.join()和onlineThread.join()表示这两个线程插队，先执行完。最后关闭executor
  • 6、executor.awaitTermination(1,TimeUnit.DAYS)防止Main线程提前退出

扩展：

  • 1、poll() 另个一个方法就是不会等待，直接去Future队列中取对象，没有返回立即null
  • 2、take() 这个方法就是一直等到拿到对象，否则一直阻塞。

3、如何在executor拒绝执行任务

类似：打烊了不接客了（。。。。。。）

当执行shutdown()并不会立即关闭，因为有线程没有执行完，如果你这时候重新请求执行新任务，应该是拒绝的。

例子：采用实现RejectedExecutionHandler来实现这个功能

package com.jack;

import java.util.concurrent.TimeUnit;

public class Task implements Runnable {

private String name;

public Task(String name) {
super();
this.name = name;
}

@Override
public void run() {
System.out.printf("Task " + name + ": 开始了\n");
try{
long duration = (long) (Math.random()*10);
System.out.printf("Task %s: ReportGenerator : 生成一个报告持续了%d 秒\n", name, duration);
TimeUnit.SECONDS.sleep(duration);
} catch (InterruptedException e){
e.printStackTrace();
}
System.out.printf("任务 %s : 结束中'''\n", name);

}

public String toString(){
return name;
}



}
package com.jack;import java.util.concurrent.RejectedExecutionHandler;import java.util.concurrent.ThreadPoolExecutor;public class RejectTaskController implements RejectedExecutionHandler {@Overridepublic void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {System.out.printf("RejectedTaskController: 任务 %s 已经被拒绝了\n", r.toString());System.out.printf("RejectedTaskController： %s\n", executor.toString());System.out.printf("RejectedTaskController: 终结中。。:%s\n", executor.toString());System.out.printf("RejectedTasksController: 终结了：%s\n", executor.isTerminated());}}

package com.jack;

import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;

public class Main {
public static void main(String[] args) {
RejectTaskController controller = new RejectTaskController();
ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newCachedThreadPool();
executor.setRejectedExecutionHandler(controller);
System.out.printf("Main : 开始中...\n");
for (int i=0; i<3; i++){
Task task = new Task("Task" +i);
executor.submit(task);
}
System.out.printf("Main: 关闭这个Executor.\n");
executor.shutdown();
System.out.printf("Main : 发送其他任务\n");
Task task = new Task("RejectedTask");
executor.submit(task);
System.out.printf("Main : 结束了\n");
}
}

日志： 
Main : 开始中...
Task Task0: 开始了
Main: 关闭这个Executor.
Task Task2: 开始了
Task Task1: 开始了
Main : 发送其他任务
RejectedTaskController: 任务 java.util.concurrent.FutureTask@7f31245a 已经被拒绝了
Task Task2: ReportGenerator : 生成一个报告持续了4 秒
Task Task1: ReportGenerator : 生成一个报告持续了3 秒
Task Task0: ReportGenerator : 生成一个报告持续了8 秒
RejectedTaskController： java.util.concurrent.ThreadPoolExecutor@6d6f6e28[Shutting down, pool size = 3, active threads = 3, queued tasks = 0, completed tasks = 0]
RejectedTaskController: 终结中。。:java.util.concurrent.ThreadPoolExecutor@6d6f6e28[Shutting down, pool size = 3, active threads = 3, queued tasks = 0, completed tasks = 0]
RejectedTasksController: 终结了：false
Main : 结束了
任务 Task1 : 结束中'''
任务 Task2 : 结束中'''
任务 Task0 : 结束中'''


总结：

  • 1、实现RejectedExecutionHandler接口的实现类RejectTaskController, 重写rejectedExecution方法
  • 2、创建执行线程的ThreadPoolExecutor executor 
  • 3、设置executor拒绝方法 setRejectedExecutionHandler();
  • 4、开启三个任务，然后关闭shutdown()方法，发现线程还在运行
  • 5、此时再创建任务让executor执行将会被拒绝了
  • 6、如果没有有拒绝处理类，将会抛出异常RejectedExecutionExeption。（它运行时异常，你可以不用捕获它）

第四章完。。。。。。。。

标签：

相关文章