定制并发类(七)实现ThreadFactory接口生成自定义的线程给Fork/Join框架

简介:

声明:本文是《 Java 7 Concurrency Cookbook 》的第七章,作者: Javier Fernández González     译者:许巧辉

实现ThreadFactory接口生成自定义的线程给Fork/Join框架

Fork/Join框架是Java7中最有趣的特征之一。它是Executor和ExecutorService接口的一个实现,允许你执行Callable和Runnable任务而不用管理这些执行线程。

这个执行者面向执行能被拆分成更小部分的任务。主要组件如下:

  • 一个特殊任务,实现ForkJoinTask类
  • 两种操作,将任务划分成子任务的fork操作和等待这些子任务结束的join操作
  • 一个算法,优化池中线程的使用的work-stealing算法。当一个任务正在等待它的子任务(结束)时,它的执行线程将执行其他任务(等待执行的任务)。

ForkJoinPool类是Fork/Join的主要类。在它的内部实现,有如下两种元素:

  • 一个存储等待执行任务的列队。
  • 一个执行任务的线程池

在这个指南中,你将学习如何实现一个在ForkJoinPool类中使用的自定义的工作者线程,及如何使用一个工厂来使用它。

准备工作…

这个指南的例子使用Eclipse IDE实现。如果你使用Eclipse或其他IDE,如NetBeans,打开它并创建一个新的Java项目。

如何做…

按以下步骤来实现的这个例子:

1.创建一个继承ForkJoinWorkerThread类的MyWorkerThread类。

public class MyWorkerThread extends ForkJoinWorkerThread {

2.声明和创建一个参数化为Integer类的ThreadLocal属性,名为taskCounter。

private static ThreadLocal<Integer> taskCounter=new ThreadLocal<Integer>();

3.实现这个类的构造器。

protected MyWorkerThread(ForkJoinPool pool) {
super(pool);
}

4.重写onStart()方法。调用父类的这个方法,写入一条信息到控制台。设置当前线程的taskCounter属性值为0。

@Override
protected void onStart() {
super.onStart();
System.out.printf("MyWorkerThread %d: Initializing task
counter.\n",getId());
taskCounter.set(0);
}

5.重写onTermination()方法。写入当前线程的taskCounter属性值到控制台。

@Override
protected void onTermination(Throwable exception) {
System.out.printf("MyWorkerThread %d:
%d\n",getId(),taskCounter.get());
super.onTermination(exception);
}

6.实现addTask()方法。递增taskCounter属性值。

public void addTask(){
int counter=taskCounter.get().intValue();
counter++;
taskCounter.set(counter);
}

7.创建一个实现ForkJoinWorkerThreadFactory接口的MyWorkerThreadFactory类。实现newThread()方法,创建和返回一个MyWorkerThread对象。

@Override
public ForkJoinWorkerThread newThread(ForkJoinPool pool) {
return new MyWorkerThread(pool);
}
}

8.创建MyRecursiveTask类,它继承一个参数化为Integer类的RecursiveTask类。

public class MyRecursiveTask extends RecursiveTask<Integer> {

9.声明一个私有的、int类型的属性array。

private int array[];

10.声明两个私有的、int类型的属性start和end。

private int start, end;

11.实现这个类的构造器,初始化它的属性。

public MyRecursiveTask(int array[],int start, int end) {
this.array=array;
this.start=start;
this.end=end;
}

12.实现compute()方法,用来合计数组中在start和end位置之间的所有元素。首先,将执行这个任务的线程转换成一个MyWorkerThread对象,然后使用addTask()方法来增长这个线程的任务计数器。

@Override
protected Integer compute() {
Integer ret;
MyWorkerThread thread=(MyWorkerThread)Thread.currentThread();
thread.addTask();
}

13.实现addResults()方法。计算和返回两个任务(接收参数)的结果的总和。

private Integer addResults(Task task1, Task task2) {
int value;
try {
value = task1.get().intValue()+task2.get().intValue();
} catch (InterruptedException e) {
e.printStackTrace();
value=0;
} catch (ExecutionException e) {
e.printStackTrace();
value=0;
}

14.令这个线程睡眠10毫秒,然后返回任务的结果。

try {
TimeUnit.MILLISECONDS.sleep(10);
} catch (InterruptedException e) {
e.printStackTrace();
}
return value;
}

15.实现这个例子的主类,通过创建Main类,并实现main()方法。

public class Main {
public static void main(String[] args) throws Exception {

16.创建一个名为factory的MyWorkerThreadFactory对象。

MyWorkerThreadFactory factory=new MyWorkerThreadFactory();

17.创建一个名为pool的ForkJoinPool对象,将前面创建的factory对象作为参数传给它的构造器。

ForkJoinPool pool=new ForkJoinPool(4, factory, null, false);

18.创建一个大小为100000的整数数组,将所有元素初始化为值1。

int array[]=new int[100000];
for (int i=0; i<array.length; i++){
array[i]=1;
}

19.创建一个新的Task对象,用来合计数组中的所有元素。

MyRecursiveTask task=new MyRecursiveTask(array,0,array.length);

20.使用execute()方法,将这个任务提交给池。

pool.execute(task);

21.使用join()方法,等待这个任务的结束。

task.join();

22.使用shutdown()方法,关闭这个池。

pool.shutdown();

23.使用awaitTermination()方法,等待这个执行者的结束。

pool.awaitTermination(1, TimeUnit.DAYS);

24.使用get()方法,将任务的结束写入到控制台。

System.out.printf("Main: Result: %d\n",task.get());

25.写入一条信息到控制台,表明程序的结束。

System.out.printf("Main: End of the program\n");

它是如何工作的…

Fork/Join框架使用的线程叫工作者线程。Java包含继承Thread类的ForkJoinWorkerThread类和使用Fork/Join框架实现工作者线程。

在这个指南中,你已实现了继承ForkJoinWorkerThread类的MyWorkerThread类,并重写这个类的两个方法。你的目标是实现每个工作者线程的任务计数器,以至于你可以知道每个工作者线程执行多少个任务。你已经通过一个ThreadLocal属性实现计数器。这样,每个线程都拥有它自己的计数器,对于来你说是透明的。

你已重写ForkJoinWorkerThread类的onStart()方法来实现任务的计数器。当工作者线程开始它的执行时,这个方法将被调用。你也重写了onTermination()方法,将任务计数器的值写入到控制台。当工作者线程结束它的执行时,这个方法将被调用。你也在MyWorkerThread类中实现addTask()方法,用来增加每个线程的任务计数器。

对于ForkJoinPool类,与Java并发API中的所有执行者一样,使用工厂来创建它。所以,如果你想在ForkJoinPool类中使用MyWorkerThread线程,你必须实现自己的线程工厂。对于Fork/Join框架,这个工厂必须实现ForkJoinPool.ForkJoinWorkerThreadFactory类。为此,你已实现MyWorkerThreadFactory类。这个类只有一个用来创建一个新的MyWorkerThread对象的方法。

最后,你只要使用已创建的工厂来初始化ForkJoinPool类。你已在Main类中通过使用ForkJoinPool的构造器实现了。

以下截图显示了这个程序的部分输出:

4

你可以看出ForkJoinPool对象如何执行4个工作者线程及每个工作者线程执行多少个任务。

不止这些…

考虑一下,当一个线程正常结束或抛出一个Exception异常时,调用的ForkJoinWorkerThread提供的onTermination()方法。这个方法接收一个Throwable对象作为参数。如果这个参数值为null时,表明这个工作者线程正常结束。但是,如果这个参数的值不为null,表明这个线程抛出一个异常。你必须包含必要的代码来处理这种情况。

参见

目录
相关文章
|
1天前
|
数据采集 存储 Java
高德地图爬虫实践:Java多线程并发处理策略
高德地图爬虫实践:Java多线程并发处理策略
|
7天前
|
安全 Java
深入理解 Java 多线程和并发工具类
【4月更文挑战第19天】本文探讨了Java多线程和并发工具类在实现高性能应用程序中的关键作用。通过继承`Thread`或实现`Runnable`创建线程,利用`Executors`管理线程池,以及使用`Semaphore`、`CountDownLatch`和`CyclicBarrier`进行线程同步。保证线程安全、实现线程协作和性能调优(如设置线程池大小、避免不必要同步)是重要环节。理解并恰当运用这些工具能提升程序效率和可靠性。
|
9天前
|
Java 开发者
Java中多线程并发控制的实现与优化
【4月更文挑战第17天】 在现代软件开发中,多线程编程已成为提升应用性能和响应能力的关键手段。特别是在Java语言中,由于其平台无关性和强大的运行时环境,多线程技术的应用尤为广泛。本文将深入探讨Java多线程的并发控制机制,包括基本的同步方法、死锁问题以及高级并发工具如java.util.concurrent包的使用。通过分析多线程环境下的竞态条件、资源争夺和线程协调问题,我们提出了一系列实现和优化策略,旨在帮助开发者构建更加健壮、高效的多线程应用。
7 0
|
13天前
|
Java API 调度
安卓多线程和并发处理:提高应用效率
【4月更文挑战第13天】本文探讨了安卓应用中多线程和并发处理的优化方法,包括使用Thread、AsyncTask、Loader、IntentService、JobScheduler、WorkManager以及线程池。此外,还介绍了RxJava和Kotlin协程作为异步编程工具。理解并恰当运用这些技术能提升应用效率,避免UI卡顿,确保良好用户体验。随着安卓技术发展,更高级的异步处理工具将助力开发者构建高性能应用。
|
17天前
|
存储 安全 Java
java多线程之原子操作类
java多线程之原子操作类
|
18天前
|
NoSQL Java Redis
Java自定义线程池的使用
Java自定义线程池的使用
|
18天前
|
Java
Java中的多线程实现:使用Thread类与Runnable接口
【4月更文挑战第8天】本文将详细介绍Java中实现多线程的两种方法:使用Thread类和实现Runnable接口。我们将通过实例代码展示如何创建和管理线程,以及如何处理线程同步问题。最后,我们将比较这两种方法的优缺点,以帮助读者在实际开发中选择合适的多线程实现方式。
23 4
|
20天前
|
Java Spring
springboot单类集中定义线程池
该内容是关于Spring中异步任务的配置和使用步骤。首先,在启动类添加`@EnableAsync`注解开启异步支持。然后,自定义线程池类`EventThreadPool`,设置核心和最大线程数、存活时间等参数。接着,将线程池bean注入到Spring中,如`@Bean(&quot;RewardThreadPool&quot;)`。最后,在需要异步执行的方法上使用`@Async`注解,例如在一个定时任务类中,使用`@Scheduled(cron = &quot;...&quot;)`和`@Async`结合实现异步定时任务。
16 2
|
25天前
|
安全 Java
Java中的多线程并发控制
在Java中,多线程是实现并发执行任务的一种重要方式。然而,随着多个线程同时访问共享资源,可能会导致数据不一致和其他并发问题。因此,了解并掌握Java中的多线程并发控制机制显得尤为重要。本文将深入探讨Java的多线程并发控制,包括synchronized关键字、Lock接口、Semaphore类以及CountDownLatch类等,并通过实例代码演示其使用方法和注意事项。
12 2
|
30天前
|
Linux API C++
【C++ 线程包裹类设计】跨平台C++线程包装类:属性设置与平台差异的全面探讨
【C++ 线程包裹类设计】跨平台C++线程包装类:属性设置与平台差异的全面探讨
51 2