youyichannel

志于道,据于德,依于仁,游于艺!

0%

Fork / Join 框架

什么是 Fork / Join 框架

Fork/Join 框架是 Java 7 提供的一个用于并行执行任务的框架,是一个把大任务分割成若干个小任务,最终汇总每个小任务结果后得到大任务结果的框架。

Fork 就是把一个大任务切分为若干子任务并行的执行,Join 就是合并这些子任务的执行结果,最后得到这个大任务的结果。分治思想。

工作窃取算法

工作窃取(work-stealing)算法是指某个线程从其他队列里窃取任务来执行。

为什么需要?

假如需要做一个比较大的任务,可以把这个任务分割为若干互不依赖的子任务,为了减少线程间的竞争,把这些子任务分别放到不同的队列里,并为每个队列创建一个单独的线程来执行队列里的任务,线程和队列一一对应。

但是,有的线程会先把自己队列里的任务干完,而其他线程对应的队列里还有任务等待处理。干完活的线程与其等着,不如去帮其他线程干活,于是它就去其他线程的队列里窃取一个任务来执行。而在这时它们会访问同一个队列,所以为了减少窃取任务线程和被窃取任务线程之间的竞争,通常会使用双端队列,被窃取任务线程永远从双端队列的头部拿任务执行,而窃取任务的线程永远从双端队列的尾部拿任务执行。

工作窃取算法的优点:充分利用线程进行并行计算,减少了线程间的竞争。

工作窃取算法的缺点:在某些情况下还是存在竞争,比如双端队列里只有一个任务时。并且该算法会消耗了更多的系统资源,比如创建多个线程和多个双端队列。

Fork / Join 框架的设计

步骤

  1. 分割任务:不停分割,直到任务足够小
  2. 执行任务并合并结果

Fork/Join 使用两个类来完成以上两件事情:

  1. ForkJoinTask:要使用 ForkJoin 框架,必须首先创建一个 ForkJoin 任务。它

    提供在任务中执行fork()join()操作的机制。通常情况下,不需要直接继承 ForkJoinTask类,只需要继承它的子类,Fork/Join 框架提供了以下两个子类。

    • RecursiveAction:用于没有返回结果的任务
    • RecursiveTask:用于有返回结果的任务
  2. ForkJoinPool:ForkJoinTask 需要通过 ForkJoinPool 来执行。任务分割出的子任务会添加到当前工作线程所维护的双端队列中,进入队列的头部。当一个工作线程的队列里暂时没有任务时,它会随机从其他工作线程的队列的尾部获取一个任务。

Fork / Join 框架的使用

需求:计算 1 + 2 + ... + 100 的结果

使用 Fork/Join 框架首先要考虑到的是如何分割任务,如果希望每个子任务最多执行两个数的相加,那么设置分割的阈值是 2,由于是 100 个数字相加,所以 Fork/Join 框 架会把这个任务 fork 成50个子任务,子任务一负责计算 1+2,子任务二负责计算 3+4……,然后再join子任务的结果。因为是有返回结果的任务,所以必须要继承RecursiveTask:

public class CountTask extends RecursiveTask<Integer> {

private static final int THRESHOLD = 2;

private final int start;
private final int end;

public CountTask(int start, int end) {
this.start = start;
this.end = end;
}

@Override
protected Integer compute() {
int sum = 0;
boolean canCompute = (end - start) <= THRESHOLD;
if (canCompute) {
for (int i = start; i <= end; i++) sum += i;
} else {
// 任务长度大于阈值,分割
int mid = (start + end) / 2;
CountTask leftTask = new CountTask(start, mid);
CountTask rightTask = new CountTask(mid + 1, end);

// 执行子任务
leftTask.fork();
rightTask.fork();

// 等待子任务执行完毕,得到结果
int leftRes = leftTask.join();
int rightRes = rightTask.join();

// 合并子任务
sum = leftRes + rightRes;
}
return sum;
}

public static void main(String[] args) throws ExecutionException, InterruptedException {
ForkJoinPool pool = new ForkJoinPool();

// 生成一个计算任务,负责计算 1 + ... + 100
CountTask countTask = new CountTask(1, 100);

ForkJoinTask<Integer> task = pool.submit(countTask);

System.out.println(task.get());
}
}

Fork/Join 框架的异常处理

ForkJoinTask 在执行的时候可能会抛出异常,但是我们没办法在主线程里直接捕获异常,所以 ForkJoinTask 提供了isCompletedAbnormally()方法来检查任务是否已经抛出异常或已经被取消了,并且可以通过 ForkJoinTask 的 getException方法获取异常。

if (task.isCompletedAbnormally()) {
System.out.println(task.getException());
}

getException 方法返回 Throwable 对象,如果任务被取消了则返回CancellationException。 如果任务没有完成或者没有抛出异常则返回 null。

SynchronousQueue

SynchronousQueue是一个不存储元素的阻塞队列。每一个put操作必须等待一个take操作,否则他不能继续添加元素。

阅读全文 »

PriorityBlockingQueue

PriorityBlockingQueue是一个支持优先级的无界阻塞队列。默认情况下元素采取自然顺序升序排列,也可以实现compareTo()方法来指定元素排序规则,或者初始化PriorityBlockingQueue时,指定构造参数Comparator来对元素进行排序。需要注意的是不能保证同优先级元素的顺序。

阅读全文 »

《MySQL是怎样运行的 —— 从跟上理解MySQL》—— 第二十章

一、事务回滚的需求

事务需要保证原子性,也就是事务中的操作要么全部完成,要么什么也不做。

事务执行到一半会出现一些情况,比如:

  • 事务执行过程中可能遇到各种错误,比如服务器本身的错误,操作系统错误,甚至是突然断电导致的错误。
  • 开发者可以在事务执行过程中手动输入ROLLBACK语句结束当前的事务的执行。

这两种情况都会导致事务执行到一半就结束,但是事务执行过程中可能已经修改了很多东西,为了保证事务的原子性,此时需要把东西改回原先的样子,这个过程就称之为回滚(rollback)。

每当要对一条记录做改动时,把回滚时所需的东西都给记下来。这些为了回滚而做的记录称之为撤销日志(undo log)。

阅读全文 »

ArrayBlockingQueue VS LinkedBlockingQueue

区别

1)锁的使用上的不同。ArrayBlockingQueue使用一把全局锁,无论是放入还是取出,都使用一把锁,读和写会互相影响。对于LinkedBlockingQueue来说,它使用两把锁,一把take锁,一把put锁,存取互不影响。

从这个角度来说,LinkedBlockingQueue更加适合高并发,有更高的吞吐量,能够在单位时间内处理更多的请求。

阅读全文 »

阻塞队列简介

阻塞队列(BlockingQueue)是一个支持两个附加操作的队列。这两个附加的操作支持阻塞的插入和移除方法。

  • 支持阻塞的插入方法:当队列满时,队列会阻塞插入元素的线程,直到队列不满
  • 支持阻塞的移除方法:当队列空时,获取元素的线程会等待队列变为非空
阅读全文 »

ConcurrentLinkedQueue

在并发编程中,有时候需要使用线程安全的队列。如果要实现一个线程安全的队列有两种方式:一种是使用阻塞算法,另一种是使用非阻塞算法。使用阻塞算法的队列可以用一把锁(入队和出对使用同一把锁)或者两把锁(入队和出队使用不同的锁)等方式来实现。非阻塞的实现方式则可以使用循环CAS的方式来实现。

阅读全文 »

业务场景:两条线程分别计算 (3 * 5)(10 + 2)的结果,结果计算完成后,将两个结果相加。

阅读全文 »