集合不安全解决办法

  1. package com.lyj.demo;
  3. import org.junit.jupiter.api.Test;
  4. import org.springframework.boot.test.context.SpringBootTest;
  6. import java.util.List;
  7. import java.util.UUID;
  8. import java.util.concurrent.CopyOnWriteArrayList;
  10. /**
  11.  * @program: java-test-demo
  12.  * @Date: 2021/8/11 7:34
  13.  * @Author: 凌兮
  14.  * @Description:
  15.  */
  16. @SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE)
  17. public class JUCTest {
  19.     /**
  20.      * List非线程安全的，底层add方法没有synchronized
  21.      * JUC：就是java.util.concurrent包
  22.      */
  23.     public static void main(String [] args) {
  24.         // 方式1：list arrayList是非线程安全集合
  25.         // 报错：Exception in thread "1" Exception in thread "2"
  26.         // Exception in thread "0" java.util.ConcurrentModificationException
  27. //        List<String> list = new ArrayList<>();
  28.         //方式2：使用vector线程安全的集合，底层add方法有synchronized，该方式比较古老，jdk1.0时就有了。
  29. //        List<String> list = new Vector<>();
  30.         // 方式3：使用Collections.synchornizedList方式创建,入参传入一个集合，
  31.         // 线程安全的,底层也是用的synchronized，该方式比较古老，jdk1.2就有了
  32. //        List<String> list = Collections.synchronizedList(new ArrayList<>());
  33.         // 方式4：使用JUC 里的CopyOnWriteArrayList,采用的是写时复制技术，底层采用lock方式
  34.         List<String> list = new CopyOnWriteArrayList<>();
  35.         for (int i = 0; i < 30; i++) {
  36.             new Thread(() -> {
  37.                 list.add(UUID.randomUUID().toString().substring(0, 8));
  38.                 System.out.println(list);
  39.             }, String.valueOf(i)).start();
  40.         }
  41.                 /** ============hashSet线程不安全测试======== */
  42.         // hashset非线程安全，报错：Exception in thread "15" Exception
  43.         // in thread "19" java.util.ConcurrentModificationException
  44. //        Set<String> set = new HashSet<>();
  45.         // 解决方式1：CopyOnWriteArraySet,线程安全的，底层用的是lock
  46.         Set<String> set = new CopyOnWriteArraySet<>();
  47. //        for (int i = 0; i < 30; i++) {
  48. //            new Thread(() -> {
  49. //                set.add(UUID.randomUUID().toString().substring(0, 8));
  50. //                System.out.println(set);
  51. //            }, String.valueOf(i)).start();
  52. //        }
  53.         /** ============hashMap线程不安全测试======== */
  54.         // hashmap线程不安全的,报错：Exception in thread "3" java.util.ConcurrentModificationException
  55. //        Map<String, String> map = new HashMap<>();
  56.         // 解决方式1：ConcurrentHashMap 线程安全的
  57.         Map<String, String> map = new ConcurrentHashMap<>();
  58.         for (int i = 0; i < 30; i++) {
  59.             String key = String.valueOf(i);
  60.             new Thread(() -> {
  61.                 map.put(key, UUID.randomUUID().toString().substring(0, 8));
  62.                 System.out.println(map);
  63.             }, String.valueOf(i)).start();
  64.         }
  66.     }
  67. }

juc的三个辅助工具类

countDown测试：

需求：5个学生都必须先离开后，才能锁门。相当于多个线程都执行完后

  1.    /**
  2.      * countDown测试 ：
  3.      * 5个学生都必须先离开后，才能锁门。相当于多个线程都执行完后，
  4.      * countdown值减为0才执行await之后的代码。
  5.      * @throws InterruptedException
  6.      */
  7.     @Test
  8.     public void countDownLatchTest() throws InterruptedException {
  10.         CountDownLatch countDownLatch = new CountDownLatch(5);
  11.         for (int i = 1; i <= 5; i++) {
  12.              new Thread(() -> {
  13.                  System.out.println(Thread.currentThread().getName() + "我离开教室了");
  14.                  countDownLatch.countDown();
  15.              }, String.valueOf(i)).start();
  16.          }
  17.          countDownLatch.await();
  18.         System.out.println(Thread.currentThread().getName() + "班长锁门了，所有人都离开教室了");
  19.     }

这个是只有当countdown值减为0才执行await之后的代码
一个线程A可使用CountDownLatch.await()方法阻塞等待，其他线程可调用CountDownLatch.countDown()方法使CountDownLatch数值减1，当CountDownLatch数值为0的时候，线程A可以继续往下执行

循环栅栏cyclicBarrier测试

需求：集齐7个龙珠召唤神龙

  1.  /**
  2.      * 循环栅栏cyclicBarrier测试
  3.      * 案例：集齐7个龙珠召唤神龙
  4.      */
  5.     @Test
  6.     public void cyclicBarrierTest() {
  7.         // 创建循环栅栏
  8.         CyclicBarrier cyclicBarrier = new CyclicBarrier(7, () -> {
  9.             // 只有parties达到7，才会执行该线程，召唤神龙
  10.             System.out.println("集齐了7个龙珠召唤神龙");
  11.         });
  12.         // 集齐龙珠过程，如果i的最大值改为6，则cyclicBarrier会一直等待，不会召唤神龙
  13.         for (int i = 1; i <= 7; i++) {
  14.             new Thread(() -> {
  15.                 System.out.println(Thread.currentThread().getName() + "龙珠");
  16.                 try {
  17.                     // 等待，parties加1，未达到7个线程时，cyclicBarrier一直会等待。
  18.                     cyclicBarrier.await();
  19.                 } catch (InterruptedException e) {
  20.                     e.printStackTrace();
  21.                 } catch (BrokenBarrierException e) {
  22.                     e.printStackTrace();
  23.                 }
  24.             }, String.valueOf(i)).start();
  25.         }
  28.     }

Semaphore信号量测试

需求：6辆汽车，3个停车位

  1. /**
  2.      * Semaphore信号量测试：
  3.      * 6辆汽车，3个停车位
  4.      */
  5.     @Test
  6.     public void semaphoreTest() {
  8.         // 创建信号量，具有3个许可证
  9.         Semaphore semaphore = new Semaphore(3);
  11.         for (int i = 1; i <= 6; i++) {
  12.             new Thread(() -> {
  13.                 try {
  14.                     // 抢占许可证
  15.                     semaphore.acquire();
  16.                     System.out.println(Thread.currentThread().getName() + "获得了许可证并停车了");
  17.                     // 设置随机停车时间
  18.                     TimeUnit.SECONDS.sleep(new Random().nextInt(5));
  19.                     System.out.println(Thread.currentThread().getName() + "归还了许可证并离开车库");
  20.                 } catch (InterruptedException e) {
  21.                     e.printStackTrace();
  22.                 } finally {
  23.                     // 释放许可证，并离开
  24.                     semaphore.release();
  25.                 }
  26.             }, String.valueOf(i)).start();
  27.         }
  28.     }

没有许可证无法执行

读写锁：

  1.      static class MapCache {
  3.         // 创建缓存器 由于是经常进行读写操作，所以设置为volatile
  4.         public static volatile Map<String, Object> cache = new HashMap<>();
  5.         // 创建读写锁
  6.         public static ReadWriteLock lock = new ReentrantReadWriteLock();
  7.         // 读锁
  8.         public static Lock readLock = lock.readLock();
  9.         // 写锁
  10.         public static Lock writeLock = lock.writeLock();
  11.         /**
  12.          * 写操作
  13.          * @param key
  14.          * @param value
  15.          */
  16.         public static void put(String key, Object value) {
  17.             writeLock.lock();
  18.             // 暂停一会
  19.             try {
  20.                 System.out.println(Thread.currentThread().getName() + "正在进行写操作" + key);
  21.                 TimeUnit.MILLISECONDS.sleep(300);
  22.                 cache.put(key, value);
  23.                 System.out.println(Thread.currentThread().getName() + "写入完毕" + key);
  24.             } catch (InterruptedException e) {
  25.                 e.printStackTrace();
  26.             } finally {
  27.                 writeLock.unlock();
  28.             }
  29.         }
  31.         /**
  32.          * 读操作
  33.          * @param key
  34.          * @return
  35.          * @throws InterruptedException
  36.          */
  37.         public static Object get(String key) {
  38.             readLock.lock();
  39.             Object result = null;
  40.             // 暂停一会
  41.             try {
  42.                 System.out.println(Thread.currentThread().getName() + "正在进行读操作" + key);
  43.                 TimeUnit.MILLISECONDS.sleep(300);
  44.                 result = cache.get(key);
  45.                 System.out.println(Thread.currentThread().getName() + "读取完毕" + key);
  46.             } catch (InterruptedException e) {
  47.                 e.printStackTrace();
  48.             } finally {
  49.                 readLock.unlock();
  50.             }
  51.             return result;
  52.         }
  53.     }
  55. // 测试
  56.  public static void main(String[] args) {
  57.         // 多线程写
  58.         for (int i = 1; i <= 5; i++) {
  59.             final int key = i;
  60.             new Thread(() -> {
  61.                 JUCTest.MapCache.put(key + "", key + "");
  62.             }, String.valueOf(i)).start();
  63.         }
  64.         // 多线程读
  65.         for (int i = 1; i <= 5; i++) {
  66.             final int key = i;
  67.             new Thread(() -> {
  68.                 JUCTest.MapCache.get(key + "");
  69.             }, String.valueOf(i)).start();
  70.         }
  71.     }

ReadWriteLock提供了一个读写锁(也称为共享锁和排他锁)的机制，多个线程可以同时获取到读锁(共享锁)，而同一时刻只能有一个线程获取到写锁(排他锁)。

阻塞队列

  1.     /**
  2.      * 阻塞队列测试1--抛出异常
  3.      */
  4.     @Test
  5.     public void blockingQueueTest() {
  7.         // 创建数组类型的阻塞队列
  8.         BlockingQueue<String> queue = new ArrayBlockingQueue<>(3);
  10.         System.out.println(queue.add("a"));
  11.         System.out.println(queue.add("b"));
  12.         System.out.println(queue.add("c"));
  13.         // 数组类型的阻塞队列，检查元素时，默认是第一个元素
  14.         System.out.println(queue.element());
  15.         // 再次添加会报错，queue full
  16. //        System.out.println(queue.add("d"));
  18.         System.out.println(queue.remove());
  19.         System.out.println(queue.remove());
  20.         System.out.println(queue.remove());
  21.         // 再次移除会报错，Nosuchemenet
  22. //        System.out.println(queue.remove());
  24.     }
  26.     /**
  27.      * 阻塞队列测试2--特殊值
  28.      */
  29.     @Test
  30.     public void blockingQueueTest2() {
  32.         // 创建数组类型的阻塞队列
  33.         BlockingQueue<String> queue = new ArrayBlockingQueue<>(3);
  35.         System.out.println(queue.offer("a"));
  36.         System.out.println(queue.offer("b"));
  37.         System.out.println(queue.offer("c"));
  38.         // 数组类型的阻塞队列，检查元素时，默认是第一个元素
  39.         System.out.println(queue.peek());
  40.         // 再次添加返回false, 添加成功返回true
  41.         System.out.println(queue.offer("d"));
  43.         System.out.println(queue.poll());
  44.         System.out.println(queue.poll());
  45.         System.out.println(queue.poll());
  46.         // 再次移除会返回null
  47.         System.out.println(queue.poll());
  48.     }
  50.     /**
  51.      * 阻塞队列测试3--阻塞
  52.      */
  53.     @Test
  54.     public void blockingQueueTest3() throws InterruptedException {
  56.         // 创建数组类型的阻塞队列
  57.         BlockingQueue<String> queue = new ArrayBlockingQueue<>(3);
  58.         queue.put("a");
  59.         queue.put("b");
  60.         queue.put("c");
  61.         // 再放程序会阻塞，直到队列空出一个位置。
  62. //        queue.put("d");
  63.         System.out.println("检测程序是否执行完");
  64.         System.out.println(queue.take());
  65.         System.out.println(queue.take());
  66.         System.out.println(queue.take());
  67.         // 再取程序会阻塞，直到队列里有值，获得值
  68.         System.out.println(queue.take());
  69.         System.out.println("检测程序是否执行完");
  70.     }
  72.     /**
  73.      * 阻塞队列测试4--超时
  74.      */
  75.     @Test
  76.     public void blockingQueueTest4() throws InterruptedException {
  78.         // 创建数组类型的阻塞队列
  79.         BlockingQueue<String> queue = new ArrayBlockingQueue<>(3);
  80.         System.out.println(queue.offer("a", 10, TimeUnit.MILLISECONDS));
  81.         System.out.println(queue.offer("b", 10, TimeUnit.MILLISECONDS));
  82.         System.out.println(queue.offer("c", 10, TimeUnit.MILLISECONDS));
  83.         // 再次添加，如果添加不进去，并且超时时间到了，会放弃添加
  84.         System.out.println(queue.offer("d", 10, TimeUnit.MILLISECONDS));
  86.         System.out.println("检测程序是否执行完");
  87.         System.out.println(queue.poll(10, TimeUnit.MILLISECONDS));
  88.         System.out.println(queue.poll(10, TimeUnit.MILLISECONDS));
  89.         System.out.println(queue.poll(10, TimeUnit.MILLISECONDS));
  90.         // 再次移除，如果没有移除的值，并且超时时间到了，会放弃移除，并返回null
  91.         System.out.println(queue.poll(10, TimeUnit.MILLISECONDS));
  92.         System.out.println("检测程序是否执行完");
  93.     }

image.png

Phaser

  1. public class TestPhaser {
  2.     static Random r = new Random();
  3.     static MarriagePhaser phaser = new MarriagePhaser();
  5.     public static void main(String[] args) throws InterruptedException {
  6.         phaser.bulkRegister(7);
  7.         for (int i = 0; i < 5; i++) {
  8.             new Person("person" + i).start();
  9.         }
  10.         new Person("新郎").start();
  11.         new Person("新娘").start();
  12.     }
  14.     static class Person extends Thread {
  15.         String name;
  17.         public Person(String name) {
  18.             this.name = name;
  19.         }
  21.         public void arrive() throws InterruptedException {
  22.             Thread.sleep(r.nextInt(1000));
  23.             System.out.printf("%s 到达现场！\n", name);
  24.             phaser.arriveAndAwaitAdvance();
  25.         }
  27.         public void eat() throws InterruptedException {
  28.             Thread.sleep(r.nextInt(1000));
  29.             System.out.printf("%s 吃完!\n", name);
  30.             phaser.arriveAndAwaitAdvance();
  31.         }
  33.         public void leave() throws InterruptedException {
  34.             if (name.equals("新郎") || name.equals("新娘")) {
  35.                 phaser.arriveAndAwaitAdvance();
  36.             } else {
  37.                 Thread.sleep(r.nextInt(1000));
  38.                 System.out.printf("%s 离开！\n", name);
  39.                 phaser.arriveAndAwaitAdvance();
  40.             }
  41.         }
  43.         private void hug() throws InterruptedException {
  44.             if (name.equals("新郎") || name.equals("新娘")) {
  45.                 Thread.sleep(r.nextInt(1000));
  46.                 System.out.printf("%s 抱抱！\n", name);
  47.                 phaser.arriveAndDeregister();
  48.             } else {
  49.                 phaser.arriveAndDeregister();
  50.             }
  51.         }
  53.         @Override
  54.         public void run() {
  55.             try {
  56.                 arrive();
  57.                 eat();
  58.                 leave();
  59.                 hug();
  60.             } catch (InterruptedException e) {
  61.                 e.printStackTrace();
  62.             }
  63.         }
  64.     }
  65. }
  67. class MarriagePhaser extends Phaser {
  68.     @Override
  69.     protected boolean onAdvance(int phase, int registeredParties) {
  70.         switch (phase) {
  71.         case 0:
  72.             System.out.println("所有人到齐");
  73.             System.out.println("----------");
  74.             return false;
  75.         case 1:
  76.             System.out.println("所有人吃完");
  77.             System.out.println("----------");
  78.             return false;
  79.         case 2:
  80.             System.out.println("所有人离开");
  81.             System.out.println("----------");
  82.             return false;
  83.         case 3:
  84.             System.out.println("婚礼结束");
  85.             System.out.println("----------");
  86.             return true;
  87.         default:
  88.             return true;
  89.         }
  90.     }
  91. }

phaser.getPhase() 初始值为0,如果全部线程到达集合点这个Phase+1,如果phaser.getPhase()达到Integer的最大值,这重新清空为0,在这里表示第几次集合了
phaser.arriveAndDeregister(); 表示这个线程到达集合点,就离开这个团体
phaser.arriveAndAwaitAdvance(); 表示这个线程在到某个达集合点,在等待其他线程
phaser.bulkRegister(friendNum); 表示这个线程在某个集合点遇到了friendNum个线程,他们要加入这个团体。
上述代码的意思是一共有7个人，5个person加上新郎新娘一共7个。当phaser.getPhase()为0时代码往下执行，每执行一个线程则加1，加到7后重新置为0，就可以继续向下执行了

Exchanger

  1.     public static void testExchanger() {
  2.         Exchanger<String> exchanger = new Exchanger<>();
  3.         new Thread(() -> {
  4.             String name = "Tom";
  5.             try {
  6.                 name = exchanger.exchange(name);
  7.             } catch (InterruptedException e) {
  8.                 e.printStackTrace();
  9.             }
  10.             System.out.println(Thread.currentThread().getName() + " " + name);
  11.         }, "t1").start();
  12.         new Thread(() -> {
  13.             String name = "Jack";
  14.             try {
  15.                 name = exchanger.exchange(name);
  16.             } catch (InterruptedException e) {
  17.                 e.printStackTrace();
  18.             }
  19.             System.out.println(Thread.currentThread().getName() + " " + name);
  20.         }, "t2").start();
  21.     }

输出结果：
t1 Jack
t2 Tom
从上面的程序可以看出，线程t1和t2内部的变量name发生了交换。Exchanger可用于两个线程之间交换数据，也是线程通信的一种方式。

LockSupport

  1.     public static void testLockSupport() throws InterruptedException {
  2.         Thread t1 = new Thread(() -> {
  3.             System.out.println("t1 start and park...");
  4.             LockSupport.park();
  5.             System.out.println("t1 continue and end...");
  6.         }, "t1");
  7.         t1.start();
  9.         Thread.sleep(3000);
  10.         System.out.println("3 second later, unpark t1");
  11.         LockSupport.unpark(t1);
  12.     }

LockSupport相比于wait()和notify()更具有灵活性，notify()是随机唤醒一个正在等待的线程，而LockSupport可以唤醒一个特定的线程。

  • 其实park/unpark的设计原理核心是“许可”：park是等待一个许可，unpark是为某线程提供一个许可。
    如果某线程A调用park，那么除非另外一个线程调用unpark(A)给A一个许可，否则线程A将阻塞在park操作上。

线程的run()方法是由java虚拟机直接调用的，如果我们没有启动线程（没有调用线程的start()方法）而是在应用代码中直接调用run()方法，那么这个线程的run()方法其实运行在当前线程（即run()方法的调用方所在的线程）之中，而不是运行在其自身的线程中，从而违背了创建线程的初衷；
1.直接调用 run 是在主线程中执行了 run，没有启动新的线程(t1)
2.使用 start 是启动新的线程(t1)，通过新的线程(t1)间接执行 run 中的代码

线程中的join()方法

演示示例：

  1. public class Main {
  2.     public static void main(String[] args) throws InterruptedException {
  4.         Counter counter = new Counter();
  6.         Thread tA = new Thread(new Runnable() {
  7.             @Override
  8.             public void run() {
  9.                 counter.printA();
  10.             }
  11.         });
  13.         Thread tB = new Thread(new Runnable() {
  14.             @Override
  15.             public void run() {
  16.                 counter.printB();
  17.             }
  18.         });
  20.         Thread tC = new Thread(new Runnable() {
  21.             @Override
  22.             public void run() {
  23.                 counter.printC();
  24.             }
  25.         });
  27.         tA.start();
  28.         tB.start();
  29.         tC.start();
  30.     }
  32.     static class Counter {
  34.         public void printA() {
  35.             try {
  36.                 Thread.currentThread().sleep(500);
  37.             } catch (InterruptedException e) {
  38.                 e.printStackTrace();
  39.             }
  40.             for (int i = 0; i < 5; i++) {
  41.                 System.out.println("A");
  42.             }
  43.         }
  45.         public void printB() {
  46.             try {
  47.                 Thread.currentThread().sleep(500);
  48.             } catch (InterruptedException e) {
  49.                 e.printStackTrace();
  50.             }
  51.             for (int i = 0; i < 5; i++) {
  52.                 System.out.println("B");
  53.             }
  54.         }
  56.         public void printC() {
  57.             try {
  58.                 Thread.currentThread().sleep(500);
  59.             } catch (InterruptedException e) {
  60.                 e.printStackTrace();
  61.             }
  62.             for (int i = 0; i < 5; i++) {
  63.                 System.out.println("C");
  64.             }
  65.         }
  66.     }
  67. }

join()方法的源码

  1.     public final synchronized void join(long millis)
  2.     throws InterruptedException {
  3.         long base = System.currentTimeMillis();
  4.         long now = 0;
  6.         if (millis < 0) {
  7.             throw new IllegalArgumentException("timeout value is negative");
  8.         }
  10.         if (millis == 0) {
  11.             while (isAlive()) {
  12.                 wait(0);
  13.             }
  14.         } else {
  15.             while (isAlive()) {
  16.                 long delay = millis - now;
  17.                 if (delay <= 0) {
  18.                     break;
  19.                 }
  20.                 wait(delay);
  21.                 now = System.currentTimeMillis() - base;
  22.             }
  23.         }
  24.     }

因此在tA.join()当中的wait(0)方法是让main线程陷入了无尽的等待中。正是因为如此，在tA.join()之前的代码都会正常从上往下执行，而在tA.join()之后的代码都随着main线程陷入等待而无法继续执行。这样便达到了网上说的 “t.join()方法会使所有线程都暂停并等待t的执行完毕后再执行”。