java高并发实战（十）——并发调试和JDK8新特性

痛定思痛。 2022-05-21 12:43 154阅读 0赞

由于之前看的容易忘记，因此特记录下来，以便学习总结与更好理解，该系列博文也是第一次记录，所有有好多不完善之处请见谅与留言指出，如果有幸大家看到该博文，希望报以参考目的看浏览，如有错误之处，谢谢大家指出与留言。

一、内容提要

 多线程调试的方法

 线程dump及分析

 JDK8对并发的新支持

– LongAdder

– CompletableFuture

– StampedLock

**二、多线程调试的方法**

多线程运行顺序不一致，有时很难重现bug

如下面例子：

01 public class UnsafeArrayList {
 02 static ArrayList al=new ArrayList();
 03 static class AddTask implements Runnable{
 04 @Override
 05 public void run() {
 06 try {
 07 Thread.sleep(100);
 08 } catch (InterruptedException e) {}
 09 for(int i=0;i<1000000;i++)
 10 al.add(new Object());
 11 }
 12 }
 13 public static void main(String[] args) throws InterruptedException {
 14 Thread t1=new Thread(new AddTask(),"t1");
 15 Thread t2=new Thread(new AddTask(),"t2");
 16 t1.start();
 17 t2.start();
 18 Thread t3=new Thread(new Runnable(){
 19 @Override
 20 public void run() {
 21 while(true){
 22 try {
 23 Thread.sleep(1000);
 24 } catch (InterruptedException e) {}
 25 }
 26 }
 27 },"t3");
 28 t3.start();
 29 }}

这是一个数组当容量不足时去扩容。在这里下个断点。下面展示出此时完整的堆栈，供查看当前线程在干什么。

![70][]

当所有现成经过这个点时，线程都会停下来，调试比较麻烦。所以通过条件端点去调试，去筛选出自己要查看的线程，在指定线程停下来。

![70 1][]

设置之后可以看出下面堆栈，这才是想看到的结果。当然经过dug跑到这里也可以。

![70 2][]

当我们看到上面堆栈显示的线程前面有两个黄色竖线，表示当前线程时被断下来的。绿色的三角形表示线程在运行着。同时我们也可以选中堆栈中的t1或t2中的任何一个，然后执行dug，则可以单独运行某一个线程，进行调试指定的线程。

![70 3][]

上面这个我们可以通过上面停止VM虚拟机，把所有线程都停下来，但他会产生不稳定性，根据情况而定，一般会在当其他线程影响该线程结果时，可以使用停止VM。默认情况是是停断线程设置（执行当前线程会断下来当前这个要停断的线程）。

下面是使用VM情况中断的堆栈情况：

![70 4][]

**三、线程Dump分析**

 jstack 3992

使用jstack工具去分析，导出当前正在执行的虚拟机下面的所有线程，查看线程都在做什么，那个在卡死状态等等去推理问题出现的情况。还有会Dump出来时发现每次Dump出来的线程总在执行一个步骤一句话，也没有任何锁的信息。说明他可能发生死循环。

 在%JAVA\_HOME%/bin

他的工具在这个路径可以找到它。

 分析死锁案例

代码就是上面给出的案例，通过上面调试方法测试出out of bound的出现原因。就是在线程t1执行到10时，数据在执行会扩容，但当t2也执行到这个地方发现当前数组也是10，他也执行扩容，但t1并不知道，也执行扩容，就出现了out of bound异常问题。

使用jstack  :cmd中执行jstack 3992可以看到当前线程的情况。线程谁在等待谁，谁在拥有那个线程。wait for可以查看里面的他在等在的是谁，等了很久。然后去查看这个东西被那个对象长时间拥有，给占用，但也不一定是死锁，但他就在等待很长时间，也说明系统有问题，然后通过通过-l命令打印出等待现象情况。

**四、JDK8对并发的新支持**

1、 LongAdder（累加器）

– 和AtomicInteger类似的使用方式

他的性能在高并发下AtomicLong性能更好。

– 在AtomicInteger上进行了热点分离

比如对等个hashMap加锁，当热点分离是把hashmap分16分，对16份分别加锁，就会尽可能避免冲突。是每一次的cas更新可能性提高，性能就所有提高。如下图：

– public void add(long x) 在原子上也有所实现，以及下面方法

– public void increment()

– public void decrement()

– public long sum()

– public long longValue()

– public int intValue()

LongAdder内部本来把原来表示一个整数的那个数字，分解成一个数组，每一个小单元，每个单元都是一个整数，当线程进来时，把数据打散到分成的多个单元格上，多线程时，每个线程对应每个单元格，这样就减少了线程的冲突，cas更新成功率有所提高。性能就会增加。但线程比较少时，非高并发下，他也做处理了，并不会无条件查分打散成数组，他内部也会维护一个base的元素，相当于原子long类型对应的数据，每当你对这个元素进行累加，当他发现有冲突的时候，就会创建cell数组，之后再会把线程映射这个cell数组，当线程多时，cell会不断增加扩大。有点自适应的思想。

![70 5][]

2、 CompletableFuture（工具类）

– 实现CompletionStage接口（大概有40多个方法）

– Java 8中对Future的增强版

– 支持流式调用 如下：

![70 6][]

– 完成后得到通知     （实现future功能）

01 public static class AskThread implements Runnable {
 02 CompletableFuture<Integer> re = null;
 03
 04 public AskThread(CompletableFuture<Integer> re) {
 05 this.re = re;
 06 }
 07
 08 @Override
 09 public void run() {
 10 int myRe = 0;
 11 try {
 12 myRe = re.get() * re.get();
 13 } catch (Exception e) {
 14 }
 15 System.out.println(myRe);
 16 }
 17 }
 18
 19 public static void main(String[] args) throws InterruptedException {
 20 final CompletableFuture<Integer> future = new CompletableFuture<>();
 21 new Thread(new AskThread(future)).start();
 22 // 模拟长时间的计算过程
 23 Thread.sleep(1000);
 24 // 告知完成结果
 25 future.complete(60);//使future更灵活 自由决定什么时候进行通知。
 26 }

– 异步执行  跟普通future很接近但更多用在函数式编程情况，支持函数式编程

01 public static Integer calc(Integer para) {
 02 try {
 03 // 模拟一个长时间的执行
 04 Thread.sleep(1000);
 05 } catch (InterruptedException e) {
 06 }
 07 return para*para;
 08 }
 09
 10 public static void main(String[] args) throws InterruptedException, ExecutionException {
 11 final CompletableFuture<Integer> future =
 12 CompletableFuture.supplyAsync(() -> calc(50));
 13 System.out.println(future.get());
 14 }

– 工厂方法

static CompletableFuture supplyAsync(Supplier supplier);
 static CompletableFuture supplyAsync(Supplier supplier, Executor executor);
 static CompletableFuture<Void> runAsync(Runnable runnable);
 static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor);

– 流式调用  通过工程方法创建实例后，可以通过流式方式对结果进行进一步操作

01 public static Integer calc(Integer para) {
 02 try {
 03 // 模拟一个长时间的执行
 04 Thread.sleep(1000);
 05 } catch (InterruptedException e) {
 06 }
 07 return para*para;
 08 }
 09
 10 public static void main(String[] args) throws InterruptedException, ExecutionException {
 11 CompletableFuture<Void> fu=CompletableFuture.supplyAsync(() -> calc(50)) //通过接口，构造实例；可看出这个接口更多倾向于函数式编程
 12 .thenApply((i)->Integer.toString(i))
 13 .thenApply((str)->"\""+str+"\"")
 14 .thenAccept(System.out::println);
 15 fu.get();
 16 }

– 组合多个CompletableFuture

public CompletableFuture thenCompose(Function<? super T, ? extends CompletionStage> fn)

01 public static Integer calc(Integer para) {
 02 return para/2;
 03 }
 04
 05 public static void main(String[] args) throws InterruptedException, ExecutionException {
 06 CompletableFuture<Void> fu =
 07 CompletableFuture.supplyAsync(() -> calc(50))
 08 .thenCompose((i)->CompletableFuture.supplyAsync(() -> calc(i)))
 09 .thenApply((str)->"\"" + str + "\"").thenAccept(System.out::println);
 10 fu.get();
 11 }

![70 7][]

这接口类主要功能是提供函数式编程，同时跟普通future比他提供了可以由开发是决定什么时候完成这个通知，去做future操作。这类跟性能感觉没有什么关系，对编码量大大减少，方便开发。

4、StampedLock

而锁分离的重要的实现就是ReadWriteLock。而StampedLock则是ReadWriteLock的一个改进。StampedLock与ReadWriteLock的区别在于，StampedLock认为读不应阻塞写，StampedLock认为当读写互斥的时候，读应该是重读，而不是不让写线程写。这样的设计解决了读多写少时，使用ReadWriteLock会产生写线程饥饿现象。

所以StampedLock是一种偏向于写线程的改进。

– 读写锁的改进（他认为读时候，不应该也阻塞写。应该通过从读操作）当读太多，写堵塞，会发生饥饿现象。

– 读不阻塞写

01 public class Point {
    02 private double x, y;
    03 private final StampedLock sl = new StampedLock();  //类似时间戳，每次操作会不断累加
    04
    05 void move(double deltaX, double deltaY) { // an exclusively locked method
    06 long stamp = sl.writeLock();
    07 try {
    08 x += deltaX;
    09 y += deltaY;
    10 } finally {
    11 sl.unlockWrite(stamp);
    12 }
    13 }
    14
    15 double distanceFromOrigin() { // A read-only method
    16 long stamp = sl.tryOptimisticRead();  //乐观读，所以事先读不会堵塞写
    17 double currentX = x, currentY = y;//读xy值，这里并不一定是一直的。
    18 if (!sl.validate(stamp)) {  //验证读的数据是否一致
    19 stamp = sl.readLock();//如果失败，也可以通过悲观读去去处理
    20 try {
    21 currentX = x;
    22 currentY = y;
    23 } finally {
    24 sl.unlockRead(stamp);
    25 }
    26 }
    27 return Math.sqrt(currentX * currentX + currentY * currentY);
    28 }
    29 }

述代码模拟了写线程和读线程， StampedLock根据stamp来查看是否互斥，写一次stamp变增加某个值

tryOptimisticRead()

就是刚刚所说的读写不互斥的情况。

每次读线程要读时，会先判断

if (!sl.validate(stamp))

validate中会先查看是否有写线程在写，然后再判断输入的值和当前的 stamp是否相同，即判断是否读线程将读到最新的数据。如果有写线程在写，或者 stamp数值不同，则返回失败。

如果判断失败，当然可以重复的尝试去读，在示例代码中，并没有让其重复尝试读，而采用的是将乐观锁退化成普通的读锁去读，这种情况就是一种悲观的读法。

stamp = sl.readLock();

（1）StampedLock的实现思想（解决上面验证失败后通过自旋锁去做处理）

– CLH自旋锁（他也使用了一种叫CLH的自旋锁）

当锁申请失败时，不会立即将读线程挂起，在锁当中会维护一个等待线程队列，所有申请锁，但是没有成功的线程都记录在这个队列中； 锁维护一个等待线程队列，所有申请锁，但是没有成功的线程都记录在这个队列中。每一个节点（一个节点代表一个线程），保存一个标记位（locked），用于判断当前线程是否已经释放锁。当一个线程试图获得锁时，取得当前等待队列的尾部节点作为其前序节点。并使用类似如下代码判断前序节点是否已经成功释放锁：while (pred.locked) \{\}不停的等待前面线程释放锁。

这个循环就是不断等前面那个结点释放锁，这样的自旋使得当前线程不会被操作系统挂起，从而提高了性能。当然他不会进行无休止的自旋，会在在若干次自旋后挂起线程。否则CPU的占有率就会很高。

![70 8][]

[70]: /images/20220521/4be2200236d6400f82a73daacaeb1b03.png
[70 1]: /images/20220521/cfd9fbe90f8d44c3b26c1cf98ef715e8.png
[70 2]: /images/20220521/de930a4a975e44c5b7d28d74795291d0.png
[70 3]: /images/20220521/250579f7078e47cfae85fa10243e10f1.png
[70 4]: /images/20220521/ed2944999fbf493780ab491576ab2ee5.png
[70 5]: /images/20220521/bdc6ce4aeca24d47b38dd55dd9757b87.png
[70 6]: /images/20220521/133987c12e8048a59fe4ec36ed1a0b9e.png
[70 7]: /images/20220521/b184c3baf0dd4e3a99c7322c495997c4.png
[70 8]: /images/20220521/2970b110fdc147cfa1ce04859df60516.png