crunAsync获取…

ForkJoin分治工作窃取ForkJoinPool与ThreadPoolExecutor使用案例不带返回值的计算--RecursiveAction带返回值的计算--RecursiveTask

Future

CompletableFuture为什么叫CompletableFuture?创建异步任务supplyAsyncrunAsync获取任务的方法

异步回调处理1.thenApply和thenApplyAsync2.thenAccept和thenAcceptAsync3.thenRun和thenRunAsync4、whenComplete和whenCompleteAsync5、handle和handleAsync

多任务组合处理thenCombine、thenAcceptBoth

和runAfterBoth

ForkJoinPool线程池最大的特点就是分叉(fork)合并(join)将一个大任务拆分成多个小任务并行执行再结合**工作窃取模式(worksteal)**提高整体的执行效率充分利用CPU资源。

工作窃取

工作窃取work-stealing是指当某个线程的任务队列中没有可执行任务的时候从其他线程的任务队列中窃取任务来执行以充分利用工作线程的计算能力减少线程由于获取不到任务而造成的空闲浪费。

在ForkJoinpool中工作任务的队列都采用双端队列Deque容器。

我们知道在通常使用队列的过程中我们都在队尾插入而在队头消费以实现FIFO。

而为了实现工作窃取。

一般我们会改成工作线程在工作队列上LIFO,而窃取其他线程的任务的时候从队列头部取获取。

示意图如下

工作线程worker1、worker2以及worker3都从taskQueue的尾部popping获取task而任务也从尾部Pushing当worker3队列中没有任务的时候就会从其他线程的队列中取stealing这样就使得worker3不会由于没有任务而空闲。

这就是工作窃取算法的基本原理。

可以想象要是不使用工作窃取算法那么我们在不断fork的过程中可能某些worker就会一直处于join的等待中。

工作窃取的思想实际实在golang协程的底层处理中也是如此。

ForkJoinPool与ThreadPoolExecutor

ForkJoinPool和ThreadPoolExecutor都实现了Executor和ExecutorService接口都可以通过构造函数设置线程数threadFactory可以查看ForkJoinPool.makeCommonPool()方法的源码查看通用线程池的构造细节。

在内部结构上我觉得两个线程池最大的区别是在工作队列的设计上如下图

ThreadPoolExecutor

ForkJoinPool最适合计算密集型任务而且最好是非阻塞任务。

使用案例

在JUC中实现Fork-join框架有两个类分别是ForkJoinPool以及提交的任务抽象类ForkJoinTask。

对于ForkJoinTask虽然有很多子类但是我们在基本的使用中都是使用了带返回值的RecursiveTask和不带返回值的RecursiveAction类。

案例实现打印50个任务序列

第一步构建要处理的printForkAction继承自RecursiveAction第二步重写compute()方法Forkjoin分治的思路体现在此start为开始任务序号en为结束任务序号设置任务数阈值threshold。

当要处理的任务序列数小于threshold直接循环遍历处理。

当要处理的任务序列数大于等于threshold将要处理的任务拆分一般都是中分构建两个新的printForkAction随后invokeAll(firstTask,

secondTask);

业务System.out.println(Thread.currentThread().getName()

i);}}

}第三步创建ForkJoinPool往里边提交printForkAction。

public

{e.printStackTrace();}pool.shutdown();}结果实现了多个线程共同完成大于任务序列号的任务。

案例计算1

1、继承RecursiveTask时泛型为指定返回值类型extends

firstTask.join()

secondTask.join();任务结果可以这里返回。

3、ForkJoinTaskLong

task

提交任务返回一个ForkJoinTask对象泛型任然是返回值类型4、Long

ans

ForkJoinPool();ForkJoinTaskLong

task

{System.out.println(-------);Long

ans

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

catch

{e.printStackTrace();}pool.shutdown();}

}class

Future表示一个可能还没有完成的异步任务的结果针对这个结果可以添加Callback以便在任务执行成功或失败后作出相应的操作。

Future接口主要包括5个方法

get方法可以当任务结束后返回一个结果如果调用时工作还没有结束则会阻塞线程直到任务执行完毕getlong

timeout,TimeUnit

unit做多等待timeout的时间就会返回结果cancelboolean

mayInterruptIfRunning方法可以用来停止一个任务如果任务可以停止通过mayInterruptIfRunning来进行判断则可以返回true,如果任务已经完成或者已经停止或者这个任务无法停止则会返回false.isDone方法判断当前方法是否完成isCancel方法判断当前方法是否取消

烧水案例

A线程调用B线程的join方法在B线程没有执行完成钱A线程一直处于阻塞状态join是实例方法需要用线程对象去调用使用join线程合并线程无法获取到合并线程的返回值即无法知道烧水线程执行的结果。

只能一直阻塞等待烧水线程结束

那么我们可以构建两个线性一个烧水一个洗碗将其加入join到主线程然后泡茶

public

{Thread.currentThread().setName(烧水线程);System.out.println(洗好水壶);System.out.println(灌好凉水);System.out.println(放在火上);Thread.sleep(SLEEP_TIME);}

catch

{e.printStackTrace();}System.out.println(水烧开了);});hThread.start();Thread

wThread

{Thread.currentThread().setName(清洗线程);System.out.println(洗茶壶);System.out.println(洗茶杯);System.out.println(拿茶叶);Thread.sleep(SLEEP_TIME);}

catch

{e.printStackTrace();}System.out.println(洗茶叶完成);});wThread.start();//

主线程

{hThread.join();wThread.join();System.out.println(泡泡茶喝);}

catch

烧水操作和洗碗操作分别构建为callable对象装配FutureTask在主线程中获取FutureTask的结果

public

HotWaterJob();FutureTaskBoolean

hotWaterTask

清洗线程);hotWaterThread.start();washThread.start();try

{Boolean

washTask.get();drinkTea(hotWaterFlag,

washFlag);}

{System.out.println(喝茶);}}static

class

{Thread.currentThread().setName(烧水线程);System.out.println(洗好水壶);System.out.println(灌好凉水);System.out.println(放在火上);Thread.sleep(SLEEP_TIME);}

catch

false;}System.out.println(水烧开了);return

true;}}static

{Thread.currentThread().setName(清洗线程);System.out.println(洗茶壶);System.out.println(洗茶杯);System.out.println(拿茶叶);Thread.sleep(SLEEP_TIME);}

catch

false;}System.out.println(洗茶叶完成);return

true;}}

FutureTask为例同为Future的实现类。

同传统的Future相比其支持流式计算、函数式编程、完成通知、自定义异常处理等很多新的特性。

由于函数式编程在java中越来越多的被使用到熟练掌握CompletableFuture对于更好的使用java

CompletableFuture字面翻译过来就是“可完成的Future”。

同传统的Future相比较CompletableFuture能够主动设置计算的结果值主动终结计算过程即completable从而在某些场景下主动结束阻塞等待。

而Future由于不能主动设置计算结果值一旦调用get()进行阻塞等待要么当计算结果产生要么超时才会返回。

下面的示例比较简单的说明了CompletableFuture是如何被主动完成的。

在下面这段代码中由于调用了complete方法所以最终的打印结果是“manual

future

CompletableFuture.supplyAsync(()-{try{Thread.sleep(1000L);return

test;}

System.out.println(future.join());创建异步任务

supplyAsync

supplyAsync是创建带有返回值的异步任务。

它有如下两个方法一个是使用默认线程池ForkJoinPool.commonPool()的方法一个是带有自定义线程池的重载方法

带返回值异步请求默认线程池

CompletableFuture.supplyAsync(()

result;});//等待任务执行完成System.out.println(结果-

cf.get());

Executors.newSingleThreadExecutor();CompletableFutureString

CompletableFuture.supplyAsync(()

result;},

executorService);//等待子任务执行完成System.out.println(结果-

cf.get());

runAsync是创建没有返回值的异步任务。

它有如下两个方法一个是使用默认线程池ForkJoinPool.commonPool()的方法一个是带有自定义线程池的重载方法

public

something....);});//等待任务执行完成System.out.println(结果-

cf.get());

Executors.newSingleThreadExecutor();CompletableFutureVoid

something....);},

executorService);//等待任务执行完成System.out.println(结果-

cf.get());

完成时返回结果值否则抛出unchecked异常。

为了更好地符合通用函数形式的使用如果完成此

CompletableFuture所涉及的计算引发异常则此方法将引发unchecked异常并将底层异常作为其原因

public

completeExceptionally(Throwable

ex)

join()方法抛出的是uncheckException异常即RuntimeException),不会强制开发者抛出get()方法抛出的是经过检查的异常ExecutionException,

InterruptedException

complete()只是对结果提交结束的一种设置并不返回任务结果。

getNow()如果完成则返回结果值或抛出任何遇到的异常否则返回给定的

异步回调处理

表示某个任务执行完成后执行的动作即回调方法会将该任务的执行结果即方法返回值作为入参传递到回调方法中带有返回值。

thenApply和thenApplyAsync区别在于使用thenApply方法时子任务与父任务使用的是同一个线程而thenApplyAsync在子任务中是另起一个线程执行任务并且thenApplyAsync可以自定义线程池默认的使用ForkJoinPool.commonPool()线程池。

public

{System.out.println(thenApplyAsync);CompletableFutureInteger

cf1

CompletableFuture.supplyAsync(()

{System.out.println(Thread.currentThread()

cf1

{System.out.println(Thread.currentThread()

cf2

result;});System.out.println(Thread.currentThread()

---main());//等待任务1执行完成System.out.println(cf1结果-

cf1.get());//等待任务2执行完成System.out.println(cf2结果-

cf2.get());

{System.out.println(thenApply);CompletableFutureInteger

cf1

CompletableFuture.supplyAsync(()

{System.out.println(Thread.currentThread()

cf1

{System.out.println(Thread.currentThread()

cf2

result;});System.out.println(Thread.currentThread()

---main());//等待任务1执行完成System.out.println(cf1结果-

cf1.get());//等待任务2执行完成System.out.println(cf2结果-

cf2.get());

thenAccep表示某个任务执行完成后执行的动作即回调方法会将该任务的执行结果即方法返回值作为入参传递到回调方法中无返回值。

thenAccept和thenAcceptAsync与thenApply和thenApplyAsync的区别在于accept无返回值只接受有返回值的future的结果自己本身无返回值

public

CompletableFuture.supplyAsync(()

{System.out.println(Thread.currentThread()

cf1

{System.out.println(Thread.currentThread()

cf2

something....);});//等待任务1执行完成System.out.println(cf1结果-

cf1.get());//等待任务2执行完成System.out.println(cf2结果-

cf2.get());

CompletableFuture.supplyAsync(()

{System.out.println(Thread.currentThread()

cf1

{System.out.println(Thread.currentThread()

cf2

something....);});//等待任务1执行完成System.out.println(cf1结果-

cf1.get());//等待任务2执行完成System.out.println(cf2结果-

cf2.get());

thenRun表示某个任务执行完成后执行的动作即回调方法无入参无返回值。

4、whenComplete和whenCompleteAsync

whenComplete是当某个任务执行完成后执行的回调方法会将执行结果或者执行期间抛出的异常传递给回调方法如果是正常执行则异常为null回调方法对应的CompletableFuture的result和该任务一致如果该任务正常执行则get方法返回执行结果如果是执行异常则get方法抛出异常。

whenComplete

whenComplete方法会传递异常而thenApply不会传递异常。

5、handle和handleAsync

跟whenComplete基本一致区别在于handle的回调方法有返回值。

public

CompletableFuture.supplyAsync(()

{System.out.println(Thread.currentThread()

cf1

{System.out.println(Thread.currentThread()

cf2

something....);System.out.println(上个任务结果

result);System.out.println(上个任务抛出异常

e);return

result2;});//等待任务2执行完成System.out.println(cf2结果-

cf2.get());

}多任务组合处理thenCombine、thenAcceptBoth

和runAfterBoth

这三个方法都是将两个CompletableFuture组合起来处理只有两个任务都正常完成时才进行下阶段任务。

thenCombine会将两个任务的执行结果作为所提供函数的参数且该方法有返回值thenAcceptBoth同样将两个任务的执行结果作为方法入参但是无返回值runAfterBoth没有入参也没有返回值。

注意两个任务中只要有一个执行异常则将该异常信息作为指定任务的执行结果。

public

CompletableFuture.supplyAsync(()

{System.out.println(Thread.currentThread()

cf1

CompletableFuture.supplyAsync(()

{System.out.println(Thread.currentThread()

cf2

{System.out.println(Thread.currentThread()

cf3

CompletableFuture.supplyAsync(()

{System.out.println(Thread.currentThread()

cf1

CompletableFuture.supplyAsync(()

{System.out.println(Thread.currentThread()

cf2

{System.out.println(Thread.currentThread()

cf3

something....);System.out.println(a

b);});System.out.println(cf3结果-

cf3.get());

CompletableFuture.supplyAsync(()

{System.out.println(Thread.currentThread()

cf1

CompletableFuture.supplyAsync(()

{System.out.println(Thread.currentThread()

cf2

{System.out.println(Thread.currentThread()

cf3

something....);});System.out.println(cf3结果-

cf3.get());

针对一个变量首先比较它的内存值与某个期望值是否相同如果相同就给它赋一个新值。

期望值举例为B,

在J.U.C下的Atomic包提供了一系列的操作简单性能高效并能保证线程安全的类去更新基本类型变量数组元素引用类型以及更新对象中的字段类型。

Atomic包下的这些类都是采用的是乐观锁策略去原子更新数据在java中则是使用CAS操作具体实现。

compareAndSetVA

期望V设置值为A即仅在当前内存中原子变量a的值为V的情况才会把变量更新为A。

AtomicInteger

System.out.println(a.compareAndSet(10,

11));

System.out.println(a.compareAndSet(100,

11));

System.out.println(a.compareAndSet(100,

12));

System.out.println(a.compareAndSet(11,

12));

其底层是调用的Unsafe类的compareAndSet()方法

Unsafe类

Java无法直接访问底层操作系统而是通过本地native方法来访问但还是留了一个后门-Unsafe类提供了一些低层次操作如直接内存访问等Unsafe类也提供了CAS操作的native方法

/**

拿对象o在内存偏移offset处的对象与expected比较如果相等则设置o.offsetx并返回true否则返回false

public

拿对象o在内存偏移offset处的long值与expected比较如果相等则设置o.offsetx

public

拿对象o在内存偏移offset处的int值与expected比较如果相等则设置o.offsetx

public

CAS锁的问题当一个线程将期望值A修改为B然后再将B改回A那么我们的CAS锁就失效了。

为解决这个问题采用AtomicStampedReference原子引用类给变量加上一个版本号当拿到变量时每次修改时版本号

1仅当版本号与初始一致时才可以修改成功这样就规避了ABA问题。

public

AtomicStampedReference(1,1000);new

Thread(()

as.getReference());System.out.println(as.compareAndSet(1,

as.getStamp(),

as.getReference());System.out.println(as.compareAndSet(2,

as.getStamp(),

as.getReference());}).start();new

Thread(()

{e.printStackTrace();}System.out.println(b1:

as.compareAndSet(1,

as.getReference());}).start();}