Java多线程(四)：集合类不安全和常用的辅助类

本文主要内容：解决多线程操作集合类不安全问题、常用的辅助类

一、集合类不安全

ArrayList

当我们使用多线程操作同一个ArrayList时会产生以下问题：java.util.ConcurrentModificationException(并发修改异常)

public class SafeArrayList {
    public static void main(String[] args) {
        List<String> list = new ArrayList<>();

        for (int i = 1; i <= 100 ; i++) {
            new Thread(() -> {
                list.add(UUID.randomUUID().toString().substring(0, 5));
                System.out.println(list);
            }, String.valueOf(i)).start();
        }

    }
}

为了解决上述问题，我们使用如下三种方式解决问题

public class SafeArrayList {
    public static void main(String[] args) {
        // 方式一：使用Vector(默认线程安全)替代ArrayList
        //List<String> list = new Vector<>();
        // 方式二：使用Collections集合类产生一个线程安全的ArrayList
        //List<String> list = Collections.synchronizedList(new ArrayList<>());
        // 方式三：使用JUC包下的CopyOnWriteArrayList产生一个线程安全的ArrayList
        // CopyOnWrite：写入时复制是计算机程序设计领域的一种优化策略
        // CopyOnWriteArrayList底层使用Lock锁比Vector底层使用Synchronized效率要高
        List<String> list = new CopyOnWriteArrayList<>();

        for (int i = 1; i <= 100 ; i++) {
            new Thread(() -> {
                list.add(UUID.randomUUID().toString().substring(0, 5));
                System.out.println(list);
            }, String.valueOf(i)).start();
        }

    }
}

Set

当我们使用多线程操作同一个HashSet时会产生以下问题：java.util.ConcurrentModificationException(并发修改异常)

public class SafeSet {
    public static void main(String[] args) {
        Set<String> set = new HashSet<>();

        for (int i = 1; i <= 100 ; i++) {
            new Thread(() -> {
                set.add(UUID.randomUUID().toString().substring(0, 5));
                System.out.println(set);
            }, String.valueOf(i)).start();
        }
    }
}

为了解决上述问题，我们使用如下二种方式解决问题

public class SafeSet {
    public static void main(String[] args) {
        // 方式一：使用Collections集合类产生一个线程安全的HashSet
        // Set<String> set = Collections.synchronizedSet(new HashSet<>());
        // 方式二：使用JUC包下的CopyOnWriteArraySet产生一个线程安全的HashSet
        Set<String> set = new CopyOnWriteArraySet<>();

        for (int i = 1; i <= 100 ; i++) {
            new Thread(() -> {
                set.add(UUID.randomUUID().toString().substring(0, 5));
                System.out.println(set);
            }, String.valueOf(i)).start();
        }
    }
}

Map

当我们使用多线程操作同一个HashMap时会产生以下问题：java.util.ConcurrentModificationException(并发修改异常)

public class SafeMap {
    public static void main(String[] args) {
        Map<String, String> map = new HashMap<>();

        for (int i = 1; i <= 100 ; i++) {
            new Thread(() -> {
                map.put(Thread.currentThread().getName(), UUID.randomUUID().toString().substring(0, 5));
            }, String.valueOf(i)).start();
        }
    }
}

为了解决上述问题，我们使用如下二种方式解决问题

public class SafeMap {
    public static void main(String[] args) {
        // 方式一：使用Collections集合类产生一个线程安全的HashMap
        // Map<String, String> map = Collections.synchronizedMap(new HashMap<>());
        // 方式二：使用JUC包下的ConcurrentHashMap产生一个线程安全的HashMap
        Map<String, String> map = new ConcurrentHashMap<>();

        for (int i = 1; i <= 100 ; i++) {
            new Thread(() -> {
                map.put(Thread.currentThread().getName(), UUID.randomUUID().toString().substring(0, 5));
            }, String.valueOf(i)).start();
        }
    }
}

二、Callable接口方式开启多线程

Callable接口方式开启多线程

public class CallableThread {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 通过RunnableFuture接口的实现类FutureTask来传参使用Callable方式产生的多线程
        FutureTask<Integer> task = new FutureTask<>(new MyThread());
        new Thread(task).start();
        // 通过FutureTask来获取Callable接口中方法的返回值
        Integer result = task.get();
        System.out.println(result);
    }
}

class MyThread implements Callable<Integer> {

    @Override
    public Integer call() throws Exception {
        System.out.println("执行了call()");
        return 512;
    }
}

使用Callable接口的一些细节问题：
- 使用Callable会有缓存：就是如果重复调用执行同一个Callable接口实现的线程，那么call()方法只会执行一次
- FutureTask的get方法可能会产生阻塞问题

三、常用的辅助类

CountDownLatch：递减计数器

public class CountDownLatchTest {
    public static void main(String[] args) throws InterruptedException {
        
        // 需求：需要6个学生(线程)都离开再执行关门方法

        CountDownLatch countDownLatch = new CountDownLatch(6); // 初始的默认数量

        for (int i = 1; i <= 6 ; i++) {
            new Thread(() -> {
                System.out.println(Thread.currentThread().getName() + "号学生离开了");
                countDownLatch.countDown();
            }, String.valueOf(i)).start();
        }

        countDownLatch.await(); //等待计数器归零，然后再向下执行
        System.out.println("关门了..........");
    }
}

countDownLatch.countDown()：数量 - 1
countDownLatch.await()：等待计数器归零，然后再向下执行

CyclicBarrier：递加计数器

public class CyclicBarrierTest {
    public static void main(String[] args) {
        
        // 需求：集齐7条龙珠(跑完7条线程)召唤神龙(执行最终线程：在CyclicBarrier声明最后线程)
        
        CyclicBarrier cyclicBarrier = new CyclicBarrier(7, () -> {
            System.out.println("召唤神龙");
        });

        for (int i = 1; i <= 7 ; i++) {
            final int temp = i;
            new Thread(() -> {
                System.out.println("第" + temp +"只龙珠被集齐");
                try {
                    cyclicBarrier.await(); // await()方法可以让当前计数器数量+1，达到7即结束
                } catch (InterruptedException | BrokenBarrierException e) {
                    e.printStackTrace();
                }
            }).start();
        }
    }
}

cyclicBarrier.await()：当前计数器数量+1，达到自定义值即结束

Semaphore：信号量，就是一组通行证

public class SemaphoreTest {
    public static void main(String[] args) {

        // 需求：有6辆车(6条线程)需要进去停车场(完成方法)，但一次最多进入3辆

        Semaphore semaphore = new Semaphore(3);

        for (int i = 1; i <= 6 ; i++) {
            new Thread(() -> {
                try {
                    semaphore.acquire();    //获取通行证
                    System.out.println(Thread.currentThread().getName() + "进入了停车场");
                    TimeUnit.SECONDS.sleep(1);
                    System.out.println(Thread.currentThread().getName() + "离开了停车场");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    semaphore.release();    // 释放通行证
                }
            }, String.valueOf(i)).start();
        }
    }
}

semaphore.acquire()：获得，假设如果已经满了，等待，等待被释放为止

semaphore.release()：释放，会将当前的信号量 - 1，然后唤醒等待的线程

作用：多个共享资源互斥的使用，并发限流，控制最大的线程数

WarlockMT

WarlockMT