多线程(八）：多线程高阶

Posted 2021-06-15 头发都哪去了

多线程(八）：多线程高阶

• 乐观锁和悲观锁
• • 乐观锁
  • CAS(实现乐观锁的一种机制)
  • 悲观锁
• 共享锁和非共享锁
• • 共享锁
  • 非共享锁
• 公平锁和非公平锁
• • 公平锁：
  • 非公平锁：
• 自旋锁
• 可重入锁
• synchronized 锁优化（锁消除）
• JUC 常用类（`java.util.concurrent`）
• • ReentrantLock
  • 信号量
  • 计数器
  • 循环屏障
• HashMap 线程非安全
• • HashMap 线程安全方案：ConcurrentHashMap
  • Hashtable 线程安全的容器

乐观锁和悲观锁

乐观锁

乐观锁假设认为数据一般情况下不会产生并发冲突，所以在数据进行提交更新的时候，才会正式对数据是否产生并发冲突进行检测，如果发现并发冲突了，则返回用户错误的信息，让用户决定如何去做。

CAS(实现乐观锁的一种机制)

CAS: 全称 Compare and swap ，字面意思:”比较并交换“。

假设内存中的原数据V，旧的预期值A，需要修改的新值B。

1. 比较 A 与 V 是否相等。（比 较）
2. 如果比较相等，将 B 写入 V。（交换）
3. 返回操作是否成功。

当多个线程同时对某个资源进行CAS操作，只能有一个线程操作成功，但是并不会阻塞其他线程,其他线程只会收到操作失败的信号。可见 CAS 其实是一个乐观锁。

CAS 底层实现原理
Java 层面 CAS 的实现是 UnSafe 类，UnSafe 类调用了 C++ 的本地方法，通过调用操作系统的 Atomic::cmpxchg（原子指令）来实现 CAS 操作。

线程的解决方案

1. 加锁
2. ThreadLocal
3. Atomic *（乐观锁实现）

乐观锁的实现（Atomic*）：
我们使用之前线程不安全的示例代码：

public class ThreadDemo90 {
    private static int count = 0;
    //最大循环次数
    private static final int MAXSIZE = 100000;

    public static void main(String[] args) throws InterruptedException {
        Thread t1 = new Thread(new Runnable() {
            @Override
            public void run() {
                for (int i = 0; i < MAXSIZE; i++) {
                    count++;
                }
            }
        }, "t1");
        t1.start();

        Thread t2 = new Thread(new Runnable() {
            @Override
            public void run() {
                for (int i = 0; i < MAXSIZE; i++) {
                    count--;
                }
            }
        }, "t2");
        t2.start();
        t1.join();
        t2.join();

        System.out.println(count);
    }

}

该代码的执行结果为：

这个线程是非安全的，我们使用 Atomic* 来改造。代码如下：

public class ThreadDemo91 {
    private static AtomicInteger count = new AtomicInteger(0);
    //最大循环次数
    private static final int MAXSIZE = 100000;
    public static void main(String[] args) throws InterruptedException {
        Thread t1 = new Thread(new Runnable() {
            @Override
            public void run() {
                for (int i = 0; i < MAXSIZE; i++) {
                    count.getAndIncrement();// count++;
                }
            }
        }, "t1");
        t1.start();

        Thread t2 = new Thread(new Runnable() {
            @Override
            public void run() {
                for (int i = 0; i < MAXSIZE; i++) {
                    count.getAndDecrement();// count--;
                }
            }
        }, "t2");
        t2.start();
        t1.join();
        t2.join();
        System.out.println(count);
    }
}

该代码的执行结果如下：

我们发现：此程序是线程安全的。

乐观锁的性能比较高，但是存在 ABA问题。

ABA 的问题：
一个值从A变成了B又变成了A，而这个期间我们不清楚这个过程。
解决方法：
加入版本信息，例如携带 AtomicStampedReference 之类的时间戳作为版本信息，保证不会出现老的值。

AtomicStampedReference 和 AtomicReference 有什么区别?
AtomicReference 不会产生 ABA 问题；
AtomicStampedReference 会产生 ABA 问题。

乐观锁的问题：并不总是能处理所有问题，所以会引入一定的系统复杂度。

悲观锁

总是假设最坏的情况，每次去拿数据的时候都认为别人会修改，所以每次在拿数据的时候都会上锁，这样别人想拿这个数据就会阻塞直到它拿到锁。
悲观锁的问题：总是需要竞争锁，进而导致发生线程切换，挂起其他线程；所以性能不高。
例如：synchronized。

如何理解乐观锁和悲观锁，具体怎么实现呢
乐观锁：CAS 实现锁机制，CAS 是由V（内存值）A（预期旧值）B（新值）组成，然后执行的时候是用 V 和 A 对比，如果结果为 true 则表明没有并发冲突，则可以直接修改，否则不能修改。CAS 是通过调用 C++ 实现提供的 UnSafe 中的本地方法（CompareAndSwapXXX）来实现的，C++ 是通过调用操作系统 Atomic::cmpxchg（原子指令）来实现的。
悲观锁：synchronized 在 Java 中将锁的 ID 存放到对象头来实现的，synchronized 在JVM 层面是通过监视器锁来实现的，synchronized 在操作系统层面是通过互斥锁 mutex 实现。

共享锁和非共享锁

共享锁

一把锁可以被多个线程拥有。
读写锁中的读锁就是共享锁。

读写锁：
将一把锁分成两个，一个用于读数据的锁（也叫做读锁），另一个锁叫做写锁，读锁可以被多个线程同时拥有，而写锁只能被一个线程拥有。
读写锁的优势：
锁的粒度更加小，性能更高。

非共享锁

一把锁只能被一个线程拥有，例如：synchronized。

示例代码如下：

public class ThreadDemo96 {
    public static void main(String[] args) {
        ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock();

        //读锁
        ReentrantReadWriteLock.ReadLock readLock = readWriteLock.readLock();

        //写锁
        ReentrantReadWriteLock.WriteLock writeLock = readWriteLock.writeLock();

        //线程池
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                10, 10, 0, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(1000));

        //任务1：执行读锁
        executor.execute(new Runnable() {
            @Override
            public void run() {
                //加锁
                readLock.lock();
                try {
                    System.out.println(Thread.currentThread().getName() +
                            "执行了读锁操作" + new Date());
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    readLock.unlock();
                }
            }
        });
        //任务2：执行读锁
        executor.execute(new Runnable() {
            @Override
            public void run() {
                //加锁
                readLock.lock();
                try {
                    System.out.println(Thread.currentThread().getName() +
                            "执行了读锁操作" + new Date());
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    readLock.unlock();
                }
            }
        });

        //任务3：执行写锁
        executor.execute(new Runnable() {
            @Override
            public void run() {
                //加锁
                writeLock.lock();
                try {
                    System.out.println(Thread.currentThread().getName() +
                            "执行了写锁操作" + new Date());
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    writeLock.unlock();
                }
            }
        });

        //任务4：执行写锁
        executor.execute(new Runnable() {
            @Override
            public void run() {
                //加锁
                writeLock.lock();
                try {
                    System.out.println(Thread.currentThread().getName() +
                            "执行了写锁操作" + new Date());
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    writeLock.unlock();
                }
            }
        });
    }
}

该代码的执行结果如下：

可见，读锁为共享锁；写锁为非共享锁。

注意：读写锁中读锁和写锁是互斥的。
防止同时读写产生脏数据，所以读写锁的读锁和写锁是互斥的。

公平锁和非公平锁

公平锁：

锁的获取顺序和线程方法的先后顺序是一致的。
公平锁: new ReentrantLock(true)
优点：执行是有序的，所以结果也是可以预期的。

非公平锁：

锁的获取顺序和线程方法的先后顺序无关（默认锁策略）。
非公平锁: new ReentrantLock()/new ReentrantLock(false)/synchronized
优点：性能更高

自旋锁

通过死循环，一直尝试获取锁。

自旋锁的缺点:如果发生死锁则会一直自旋(循环)，所以会带来一定的额外开销。

可重入锁

示例代码如下：

public class ThreadDemo97 {
    // 创建锁
    private static Object lock = new Object();

    public static void main(String[] args) {
        // 第一次进入锁
        synchronized (lock) {
            System.out.println("第一次进入锁");
            synchronized (lock) {
                System.out.println("第二次进入锁");
            }
        }
    }
}

该代码的执行结果如下：

可见，synchronized 是可重入锁。

synchronized 锁优化（锁消除）

JDK 1.6 锁升级的过程：

JUC 常用类（java.util.concurrent）

ReentrantLock

注意事项：

1. lock() 要写在 try{} 之前。
2. 一定记得在 finally{} 里面进行 unlock()

信号量

用来控制锁的数量

信号量演示程序，代码背景：四辆车，停入两个停车位，示例代码如下：

/*
 * 信号量演示程序
 * 背景：四辆车，停入两个停车位
 */
public class ThreadDemo98 {
    public static void main(String[] args) {
        //创建信号量（车位数量）
        Semaphore semaphore = new Semaphore(2, true);

        ThreadPoolExecutor executor =
                new ThreadPoolExecutor(10, 10, 0, TimeUnit.SECONDS,
                        new LinkedBlockingQueue<>(100));

        //创建四辆车
        for (int i = 0; i < 4; i++) {
            //创建任务
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    System.out.println(Thread.currentThread().getName() + "到达停车场");
                    try {
                        Thread.sleep(1000);//确保四人到达停车场
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    //试图进入停车场
                    try {
                        //尝试获取锁
                        semaphore.acquire();
                        //代码执行到此处，说明已经获取到锁
                        System.out.println(Thread.currentThread().getName() + "进入停车场");
                        //构建车辆停留时间
                        int num = 1 + new Random().nextInt(5);
                        try {
                            Thread.sleep(num * 1000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        //代码执行到此处，说明已经获取到锁
                        System.out.println(Thread.currentThread().getName() + "离开停车场");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally {
                        semaphore.release();//释放锁
                    }
                }
            });
        }
    }
}

代码的执行效果如下：

计数器

计数器是用来保证一组线程同时完成某个操作之后，才能继续后面的任务。

CountDownLatch 的实现原理：
在 CountDownLatch 里面有一个计数器，每次调用CountDown()方法的时候，计数器的数量 -1 ，直到减到 0 之后，就可以执行 await()之后的代码了。

计数器演示程序，代码背景：五人赛跑，示例代码如下：

/*
 * 计数器示例
 * 背景：5人赛跑
 */
public class ThreadDemo99 {
    public static void main(String[] args) throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(5);
        for (int i = 1; i < 6; i++) {
            final int finalI = i;
            new Thread(new Runnable() {
                @Override
                public void run() {
                    System.out.println(Thread.currentThread().getName() + "开始起跑~");
                    try {
                        Thread.sleep(1000 * finalI);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println(Thread.currentThread().getName() + "到达终点");
                    //计数器 -1
                    latch.countDown();
                }
            }).start();
        }
        //阻塞等待
        latch.await();
        System.out.println("所有人都到达终点了~~~");
    }
}

CountDownLatch缺点:
CountDownLatch 计时器的使用是一次性的，当用完一次之后，就不能再使用了。

循环屏障

循环屏障演示程序，示例代码如下：

/*
 * 循环屏障示例
 */
public class ThreadDemo100 {
    public static void main(String[] args) {
        CyclicBarrier cyclicBarrier = new CyclicBarrier(2, new Runnable() {
            @Override
            public void run() {
                System.out.println("执行了 CyclicBarrier 里面的 Runnable");
            }
        });

        for (int i = 1; i < 5; i++) {
            int finalI = i;
            new Thread(new Runnable() {
                @Override
                public void run() {
                    System.out.println(Thread.currentThread().以上是关于多线程(八）：多线程高阶的主要内容，如果未能解决你的问题，请参考以下文章
 
 多线程(八）：多线程高阶
 八.多进程与多线程
 Java_14:多线程高阶
 多线程问题(算法高阶多线程算法)存在重复元素 II(数组哈希表)计数质数(数组数学)
 Python高阶（一） - 单线程、多线程和多进程的效率对比测试
 Java(高阶)——线程安全