线程基础知识复习
大神：Doug Lea
java.util.concurrent
java.util.concurrent.aomic
Java.util.concurrent.locks
硬件
摩尔定律：
它是由英特尔创始人之一 Gordon Moore(戈登·摩尔)提出来的。其内容：
当价格不变是，集成电路上可容纳的元器件的数目每隔18-24个月会增加一倍，性能也将提升一倍
换言之，每一美元所能买到的电脑性能，将每隔18-24个月翻一倍以上。这一定路揭示了信息技术进步的速度
可是从2003年开始cpu主频不再更新，而是采用多核而不是更快的主频
在主频不再提高而且核数不断增加的情况下，要想让程序更快就要用到并行或兵法编程
start线程
private native void start0();
java语言本身底层语言是C++
底层C++源码解读
thread.c
java线程是通过start方法启动执行的，主要内容在native方法start0中
openjdk的写JNI一般是一一对应的，Thread.java对应的就是Thread.c
start0其实就是JVM_StartThread.此时查看源代码可以看到jvm.h中找到声明，jvm.cpp中有实现
jvm.cpp
调用操作系统
thread.cpp
调用操作系统
多线程概念
1把锁
synchronized
2个并
并发 concurrent
同一实体，同一机器，同时处理多个任务
同一时刻，只有一个事件发生
并行 parallel
不同实体，多台机器，同时处理多个任务
同一时刻，多个任务互不干扰
3 个程
进程：系统中运行的一个应用程序
线程：轻量级进程，在同一个进程会有1个或多个线程
管程：monitor一种同步机制
同步指令
java虚拟机可以支持方法级的同步和方法内部一段指令列的同步，这两种同步结构都是使用管程（Monitor，更常见的是直接将它称为锁）来实现
方法级的同步是隐式的，无须通过字节码指令来控制，它实现在方法调用和返回操作之中。虚拟机可以从方法常量池中的方法表结构中的ACC_SYNCHRONIZED访问标志得知一个方法是否被声明为同步方法。当方法调用时，调用指令将会检查方法ACC_SYNCHRONIZED访问标志是否被设置，如果设置了，执行线程就要求先成功持有管程，然后才能执行方法，最后方法完成（无论是正常完成还是非正常完成）时释放管程。在方法执行期间，执行线程池有了管程，其他任何线程都无法在获取到同一个管程。如果一个同步方法执行期间抛出了异常，并且在方法内部无法处理此异常，那这个同步方法所持有的管程将在异常抛到同步方法边界之外时自动释放
同步一段指令集序列通常是由java语言的synchronized语句快来表示的，java虚拟机的指令集中有monitorenter和nonitorexit两条指令来支持synchronized关键字的语义，正确实现synchronized关键字需要javac编译器与java虚拟机两者共同协调支持
用户线程和守护线程
java线程范围两种：用户线程和守护线程
一般情况下不做任何特殊说明配置，默认都是用户线程
用户线程：系统工作线程
守护线程：是一种特殊的线程为其他线程服务的，在后台默默的完成一些系统性的服务
比如垃圾回收线程就是一个最典型的例子
判断是否是守护线程：isDaemon()
设置线程类型：setDemon()
总结
如果用户线程全部结束意味着程序需要完成的业务操作已经结束，守护线程随着JVM一同结束工作
setDaemon(true) 方法必须在start() 之前设置，否则IllegalThreadStateException异常
CompletableFuture
Future
Future 接口定义了操作异步任务执行的一些方法，如获取异步任务的执行结果、取消任务执行、判断任务是否取消、判断任务是否完毕等。
常用FutureTask异步任务
java5新加的一个接口，它提供了一种一步并行计算的功能
public static void main(String[] args) throws ExecutionException, InterruptedException {
FutureTask<String> task = new FutureTask<>(() -> {
Thread.sleep(5000);
return "hello";
});
Thread thread = new Thread(task, "task");
thread.start();
String s = task.get();
System.out.println(s);
}
优缺点
public class ThreadTest {
public static void main(String[] args) throws ExecutionException, InterruptedException {
/*long start = System.currentTimeMillis();
TimeUnit.MILLISECONDS.sleep(500);
TimeUnit.MILLISECONDS.sleep(200);
TimeUnit.MILLISECONDS.sleep(300);
long stop = System.currentTimeMillis();
long time = stop - start;
System.out.println("耗时"+time);*/
ExecutorService threadPool = Executors.newFixedThreadPool(3);
long start = System.currentTimeMillis();
FutureTask<String> f1 = new FutureTask<String>(() -> {
try {
TimeUnit.MILLISECONDS.sleep(5000);
return "f1";
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
});
threadPool.submit(f1);
FutureTask<String> f2 = new FutureTask<String>(() -> {
try {
TimeUnit.MILLISECONDS.sleep(2000);
return "f2";
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
});
threadPool.submit(f2);
FutureTask<String> f3 = new FutureTask<String>(() -> {
try {
TimeUnit.MILLISECONDS.sleep(3000);
return "f3";
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
});
threadPool.submit(f3);
TimeUnit.MILLISECONDS.sleep(500);
long stop = System.currentTimeMillis();
long time = stop - start;
// f1.get() // 过时抛出异常
System.out.println("耗时" + time + " "+f1.get()+" "+f2.get()+" "+f2.get()); // get操作 这样会导致程序阻塞，一般放在祝线程之后
System.out.println("main 忙其他任务");
threadPool.shutdown();
// 放在最后处理：while true：不断询问，isDone，如果运行完拿到结果，否则，等待两秒
}
}
缺点：耗时cpu资源（Future对与结果的获取不是很友好，需要轮询和询问）
想要复杂任务
简单任务Future完全ok
回调通知
创建异步任务（Future+线程池）
多任务前后依赖可以组合使用
CompletableFuture
为什么会出现？
get（）方法在Future计算完成之前会一直处于阻塞状态下
isDone（）方法容易耗费cpu资源
对于真正异步处理我们希望可以通过传入回调函数，在Future结束时自动调用该回调函数，这样，我们就不用等待结果
阻塞的方式和异步编程的设计理念相违背，而轮询的方式会耗费无谓的cpu资源。因此
JDK8设计出CompletableFuture
源码
CompletableFuture提供了一种观察者模式类似的机制，可以让任务执行完成之后通知监听一方
CompetionStage
-
CompletionStage代表异步计算过程中的某个阶段，一个阶段完成之后可能会触发另外一个阶段
-
一个阶段的计算可以是Function，Consume或者Runnable。比如
stage.thenApply(x->square(x))
.thenAccept(x->System.out.print(x))
.thenRun(()->Sysem.out.println())
-
一个阶段的执行可能是被单个阶段的完成触发，也可能是由多个阶段一起触发
代表异步计算过程中的某一个阶段，一个阶段完成以后可能会触发另外一个阶段，有些类似Linux系统的管道分隔符传参数
CompletableFuture
- 在java8中，CompletableFuture提供了非常强大的Future的扩展功能，可以帮助我们简化异步编程的复杂性，并且提供了函数式编程的能力，可以通过回调的方式处理计算结果，也提供了转换和组合CompletableFuture的方法
- 他可能代表一个明确完成Future，也可能代表一个完成阶段（CompletionStage），它支持在计算完成以后触发一些函数或执行某些动作
- 它实现了Future和CompletionStage接口
使用
CompletableFuture不推荐使用空参数构造使用
使用静态方法：
runAsync 无返回值
supplyAsync 有返回值
没有指定Executor的方法，直接使用默认的ForkJoinPool.commonPool()作为他线程池执行的异步代码
如果指定线程池，则使用我们自定义的或者特被指定的线程池异步代码
减少阻塞和轮询
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService poolExecutor = Executors.newFixedThreadPool(3);
try {
// 减少阻塞和轮询
CompletableFuture<String> cf = CompletableFuture.supplyAsync(
() -> {
try {
TimeUnit.SECONDS.sleep(5);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
System.out.println(10/0);
return "hello";
}
, poolExecutor);
cf.whenComplete((s, e) -> {
if (e == null) {
System.out.println("whenComplete");
}
}).exceptionally(e -> {
System.out.println("有异常");
return null;
});
System.out.println("main 去忙其他任务");
} catch (Exception e) {
throw new RuntimeException(e);
} finally {
poolExecutor.shutdown();
}
}
get和join区别：
get：编译报异常
join：编译无异常
电商
public class CompletableFutureMallDemo {
static List<NetMall> list = Arrays.asList(
new NetMall("jd"),
new NetMall("dangdang"),
new NetMall("taobao")
);
/**
* 中规中矩查法
* @param list
* @param productName
* @return
*/
public static List<String> getPrice(List<NetMall> list, String productName) {
return list.stream()
.map(netMall ->
String.format(productName + " in %s price is %.2f",
netMall.getNetMallName(), netMall.calcPrice(productName))
).collect(Collectors.toList());
}
/**
* Stream配合CompletableFuture查法
* @param list
* @param productName
* @return
*/
public static List<String> getPriceByCompletableFuture(List<NetMall> list, String productName) {
return list.stream().map(netMall ->
CompletableFuture.supplyAsync(() -> String.format(productName + " in %s price is %.2f",
netMall.getNetMallName(), netMall.calcPrice(productName))))
.collect(Collectors.toList())
.stream()
.map(s -> s.join())
.collect(Collectors.toList());
}
public static void main(String[] args) {
long startTime1 = System.currentTimeMillis();
List<String> list1 = getPrice(list, "mysql");
list1.forEach(System.out::println);
long endTime1 = System.currentTimeMillis();
System.out.println(endTime1 - startTime1);
System.out.println("------------------------");
long startTime2 = System.currentTimeMillis();
List<String> list2 = getPriceByCompletableFuture(list, "mysql");
list1.forEach(System.out::println);
long endTime2 = System.currentTimeMillis();
System.out.println(endTime2 - startTime2);
}
}
class NetMall {
@Getter
private String netMallName;
public NetMall(String netMallName) {
this.netMallName = netMallName;
}
public double calcPrice(String productName) {
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
return ThreadLocalRandom.current().nextDouble() * 2 + productName.charAt(0);
}
}
常用方法
-
获得结果和触发计算
-
获得结果
-
get()
-
get(long timeout,TimeUnit unit)
-
join()
-
getNow(T valueifAbsent)
public static void main(String[] args) { CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> { /* try { TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { throw new RuntimeException(e); }*/ return "fsfs"; }); String hehe = cf.getNow("hehe"); System.out.println(hehe); }
-
-
主动触发计算
- complete
public static void main(String[] args) { CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> { /*try { TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { throw new RuntimeException(e); }*/ return "fsfs"; }); // String hehe = cf.getNow("hehe"); // System.out.println(hehe); try { TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { throw new RuntimeException(e); } // 如果此时完成计算 false 计算的值 // 否则 true xxxxx System.out.println(cf.complete("xxxxx") + " " + cf.join()); }
-
-
对计算结果进行处理
计算结果存在依赖关系，这两个线程串形化
-
thenApply
由于存在依赖关系（当前步错，不走下一步），当前步骤有异常的话就叫停
public static void main(String[] args) { ExecutorService threadPool = Executors.newFixedThreadPool(3); CompletableFuture<Integer> cf = CompletableFuture.supplyAsync(() -> { try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { throw new RuntimeException(e); } return 0; },threadPool).thenApply(num -> { num++; System.out.println(num); System.out.println(10 / 0); return num; }).thenApply(num -> { num++; System.out.println(num); return num; }).thenApply(num -> { num++; System.out.println(num); return num; }).whenComplete((num, throwable) -> { System.out.println(num); }).exceptionally(throwable -> { System.out.println("异常"); return null; }); System.out.println(Thread.currentThread().getName() + " thread"); threadPool.shutdown(); } -
handle
有异常也可以往下一步走，根据带的异常参数可以进行一步处理
public static void main(String[] args) { ExecutorService threadPool = Executors.newFixedThreadPool(3); CompletableFuture<Integer> cf = CompletableFuture.supplyAsync(() -> { try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { throw new RuntimeException(e); } return 0; },threadPool).handle((num,e) -> { num++; System.out.println(num); System.out.println(10 / 0); return num; }).handle((num,e) -> { num++; System.out.println(num); return num; }).handle((num,e)-> { num++; System.out.println(num); return num; }).whenComplete((num, throwable) -> { System.out.println(num); }).exceptionally(throwable -> { System.out.println("异常"); return null; }); System.out.println(Thread.currentThread().getName() + " thread"); threadPool.shutdown(); }
-
-
对计算结果进行消费
thenAccept：接受任务的处理结果，并消费处理，无返回结果
总结：
thenRun：无参无返回
thenAccept：有参数无返回值
thenApply：有参数有返回值
与线程池之间的关系：
-
没有传入自定义的线程池，都用默认线程池ForkJoinPool
-
传入了一个自定线程池
如果您执行第一个任务的时候，传入了一个自定义线程池
调用thenRun方法执行第二个任务时，则第二个任务和第一个任务共用同一个线程池
调用thneRunAsync执行第二个任务时，则第一个任务使用的是您自己传入的线程池，第二个任务使用的ForkJoin线程池
-
备注：有可能处理太快，系统优化原则，直接使用main线程处理
其他如：thenAccept和thenAcceptAsync，thenApply和thenApplyAsync等，他们之间的区别也是同理
-
-
对计算速度进行选用
applyToEither：谁执行的快取谁的结果
public static void main(String[] args) { ExecutorService threadPool = Executors.newFixedThreadPool(3); CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> { try { TimeUnit.SECONDS.sleep(3); } catch (InterruptedException e) { throw new RuntimeException(e); } return "hello"; },threadPool); CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> { try { TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { throw new RuntimeException(e); } return "word"; },threadPool); CompletableFuture<String> future = f1.applyToEither(f2, f -> { return f + " is winer"; }); String join = future.join(); System.out.println(join); threadPool.shutdown(); } -
对计算结果进行合并
thenCombine
两个CompletionStage任务都完成后，最终能把两个任务的结果一起交个thenCombine来处理先完成的等着，等待其他分支任务
public static void main(String[] args) { ExecutorService threadPool = Executors.newFixedThreadPool(3); CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> { try { TimeUnit.SECONDS.sleep(3); } catch (InterruptedException e) { throw new RuntimeException(e); } return "hello"; },threadPool); CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> { try { TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { throw new RuntimeException(e); } return "world"; },threadPool); CompletableFuture<String> future = f1.thenCombine(f2, (x, y) -> x + y); String join = future.join(); System.out.println(join); threadPool.shutdown(); }
