一、业务逻辑中的并发问题

1. 示例

当存在 一个类中 的 两个方法 同时被 多个线程 执行操作 共享资源 时，需要考虑加锁。
示例如下：

public class LockTest_1 {

    private static final Logger log = LoggerFactory.getLogger(LockTest_1.class);

    volatile int a = 1;
    volatile int b = 1;

    public void add() {
        log.info("add start");
        for (int i = 0; i < 10_0000; i++) {
            a++;
            b++;
        }
        log.info("add end");
    }

    public void compare() {
        log.info("compare start");
        for (int i = 0; i < 10_0000; i++) {
            // a 始终等于 b 吗？
            // 比较操作不是原子性的，在字节码层面是会先加载 a 再加载 b 后进行比对大小
            // 当加载完a后，到b被加载时 这之间 b可能被add()又++了多次，出现了a < b的情况
            if (a < b) {
                log.info("a:{},b:{},{}", a, b, a > b);
                // 最后的 a > b 始终是 false 吗?
            }
        }
        log.info("compare start");
    }

    public static void main(String[] args) {
        LockTest_1 lockTest_1 = new LockTest_1();
        new Thread(() -> lockTest_1.add()).start();
        new Thread(() -> lockTest_1.compare()).start();
    }

}

输出结果：

2021-05-01 18:31:54.688 [INFO ] [Thread-1] [c.j.test.locktest.LockTest_1  ] compare start
2021-05-01 18:31:54.688 [INFO ] [Thread-0] [c.j.test.locktest.LockTest_1  ] add start
2021-05-01 18:31:54.695 [INFO ] [Thread-1] [c.j.test.locktest.LockTest_1  ] a:390,b:1025,false
2021-05-01 18:31:54.698 [INFO ] [Thread-1] [c.j.test.locktest.LockTest_1  ] a:52277,b:52294,true
2021-05-01 18:31:54.698 [INFO ] [Thread-1] [c.j.test.locktest.LockTest_1  ] a:56056,b:56061,false
2021-05-01 18:31:54.699 [INFO ] [Thread-1] [c.j.test.locktest.LockTest_1  ] a:62865,b:62870,false
2021-05-01 18:31:54.699 [INFO ] [Thread-1] [c.j.test.locktest.LockTest_1  ] a:64628,b:64634,true
2021-05-01 18:31:54.699 [INFO ] [Thread-1] [c.j.test.locktest.LockTest_1  ] a:71524,b:71535,false
2021-05-01 18:31:54.699 [INFO ] [Thread-1] [c.j.test.locktest.LockTest_1  ] a:78008,b:78017,false
2021-05-01 18:31:54.699 [INFO ] [Thread-1] [c.j.test.locktest.LockTest_1  ] a:83293,b:83298,true
2021-05-01 18:31:54.699 [INFO ] [Thread-1] [c.j.test.locktest.LockTest_1  ] a:87629,b:87643,true
2021-05-01 18:31:54.699 [INFO ] [Thread-1] [c.j.test.locktest.LockTest_1  ] a:93140,b:93151,true
2021-05-01 18:31:54.699 [INFO ] [Thread-0] [c.j.test.locktest.LockTest_1  ] add end
2021-05-01 18:31:54.700 [INFO ] [Thread-1] [c.j.test.locktest.LockTest_1  ] compare start

如上示例，若不加锁，两个线程同时执行 add 和 compare 方法，则 compare 会在 add 方法对 a 和 b 进行++操作时执行，并且 compare 中的比较操作也不是原子性的，底层（字节码）会先加载 a 再加载 b 最后进行比较，而 a 加载完到 b 加载这段时间，b 已经加到比 a 大了。

解决办法是两个方法都加上 synchronized，即不让两个方法同时被执行。

只对add方法加锁是没用的，因为一个类中的 同步方法 与 非同步方法 可以同时执行。

2. 指令重排

为什么 a > b ？
这是因为CPU有指令重排的机制。
指令重排序是指编译器或处理器为了优化性能而采取的一种手段，在不存在数据依赖性情况下（如写后读，读后写，写后写），调整代码执行顺序。
也就是说上面代码中，a++,b++的执行顺序可能被打乱（a、b间不存在依赖关系）

二、加锁前要清楚锁和被保护的对象是不是一个层面的

1. 示例

锁的位置加对了之后还要理清锁和要保护的对象是否是一个层面的

非静态同步方法是锁定类的实例的，而静态同步方法是锁定类的

示例如下:

public class LockTest_2 {

    public static void main(String[] args) {

        // 测试1. 多线程循环一定次数 调用Data类不同实例的add方法
        IntStream.rangeClosed(1, 10_0000)
                .parallel() // 并行流转换
                .forEach(i -> new Data().add());

        System.out.println("new十万个Data对象调用add后: " + Data.getCounter());

        // 测试2. 多线程循环一定次数 调用Data1类不同实例的add方法
        IntStream.rangeClosed(1, 10_0000)
                .parallel() // 并行流转换
                .forEach(i -> new Data1().add());

        System.out.println("new十万个Data1对象调用add后: " + Data1.getCounter());
        
        // 测试3. 多线程循环一定次数 调用Data2类不同实例的add方法
        IntStream.rangeClosed(1, 10_0000)
                .parallel() // 并行流转换
                .forEach(i -> new Data2().add());

        System.out.println("new十万个Data2对象调用add后: " + Data2.getCounter());
    }

}

class Data {

    private static int counter = 0;

    // 在非静态方法上加锁，锁定的是当前对象，
    // 这时多个对象还是共享静态变量counter，仍然有线程安全问题
    public synchronized void add() {
        counter++;
    }

    public static int getCounter() {
        return counter;
    }
}

class Data1 {

    private static int counter = 0;

    private static Object locker = new Object();

    // 对静态属性locker加锁,该类的所有实例锁定的对象都是同一个
    // 也就是该类的所有对象用的都是同一把锁
    public void add() {
        synchronized (locker) {
            counter++;
        }
    }

    public static int getCounter() {
        return counter;
    }
}

class Data2 {
    private static int counter = 0;

    // 在该静态方法上加synchronized,锁定的是class,所有实例的class都是相同的
    public synchronized static void add() {
        counter++;
    }

    public static int getCounter() {
        return counter;
    }
}

运行结果:

new十万个Data对象调用add后: 33260
new十万个Data1对象调用add后: 100000
new十万个Data2对象调用add后: 100000

测试1中，在add方法上加synchronized，锁定的是this当前实例，而add方法操作的counter静态属性是所有实例共享的。也就是说当有其他线程创建了实例后也能直接获取当前实例的锁，操作counter。这样其实add方法没有被同步，输出的肯定小于十万。

测试2中，对静态属性locker加锁，也就是该类的所有对象用的都是同一把锁。就算有多个实例调用add，同一时间也只有一个实例能拿到锁，这样就实现了同步。

测试3中，把add方法变为static方法，锁定的是class，所有实例的class都是相同的，也能有同样效果。不过这样就改变了原有的代码结构，不建议这么做。

2. 代码块级别的 synchronized 和方法上标记 synchronized 关键字，在实现上有什么区别？

他们的底层实现其实都一样，在进入同步代码之前先获取锁，获取到锁之后锁的计数器+1，同步代码执行完锁的计数器-1，如果获取失败就阻塞式等待锁的释放。
只是他们在同步块识别方式上有所不一样，从class字节码文件可以表现出来，一个是通过方法flags标志，一个是monitorenter和monitorexit指令操作。

class目录下执行 javap -c -s -v -l class名 查看字节码信息
同步方法是：flags里面多了一个ACC_SYNCHRONIZED标志，这标志用来告诉JVM这是一个同步方法，在进入该方法之前先获取相应的锁，锁的计数器加1，方法结束后计数器-1，如果获取失败就阻塞住，直到该锁被释放。
如图：

同步方法.png

同步块是：由 monitorenter 指令进入，然后 monitorexit 释放锁，在执行 monitorenter 之前需要尝试获取锁，如果这个对象没有被锁定，或者当前线程已经拥有了这个对象的锁，那么就把锁的计数器加1。当执行 monitorexit 指令时，锁的计数器也会减1。当获取锁失败时会被阻塞，一直等待锁被释放。
如图：

同步块.png

两者的本质都是对对象监视器 monitor 的获取。

详情参考：https://cloud.tencent.com/developer/article/1465413

三、synchronized

synchronized是并发编程中最常用的锁，JDK1.6以前，synchronized的底层实现是重量级的，需要找操作系统去申请锁，这会造成synchronized效率非常低。
JDK1.6开始，官方对其进行JVM层面的优化，引入了偏向锁，自旋锁，重量级锁，来减少竞争带来的上下文切换。有了锁升级的概念。

1. 锁升级

当使用synchronized的时候，HotSpot的实现是这样的：

• 第一个线程访问某把锁时，如sync(object),先在object的对象头上面的Mark Word记录这个线程。（如果只有一个线程访问时，其实没有给这个object加锁，内部实现时只是记录这个线程ID，ID相同可直接执行） 偏向锁

• 偏向锁如果有其他线程参与竞争，就会升级为 自旋锁（轻量级锁），这时其他线程并不会回到cpu的就绪队列中，而是就在那等着占用cpu，自旋访问10次没有获得锁后，锁会再次升级。自旋操作使用CAS将对象头中的Mark Word替换为指向锁记录的指针。

• 自旋失败，大概率再次自旋也是失败，因此直接升级成 重量级锁，进行线程阻塞，减少cpu消耗。当锁升级为重量级锁后，未抢到锁的线程都会被阻塞，进入阻塞队列。

2. Mark Word

Java对象头中的 Mark Word 部分存储自身的运行时数据，例如 HashCode、GC 年龄、锁相关信息等内容。
它里面存储的数据会随着锁标志位的变化而变化。
在64位虚拟机下，Mark Word是64bit大小的，其存储结构如下所示：

锁状态	25bit	31bit	1bit	4bit	1bit	2bit
			cms_free	分代年龄	偏向锁	锁标志位
无锁		hashCode			0	01
偏向锁	ThreadID(54bit) Epoch(2bit)				1	01
轻量级锁	指向栈中锁记录的指针					00
重量级锁	指向重量级锁的指针					10
GC标记	空					11

3. 监视器（Monitor）

每一个锁都对应一个monitor对象，在HotSpot虚拟机中它是由ObjectMonitor实现的（C++实现）。
每个对象都存在着一个monitor与之关联，对象与其monitor之间的关系有存在多种实现方式，如monitor可以与对象一起创建销毁或当线程试图获取对象锁时自动生成，但当一个monitor被某个线程持有后，它便处于锁定状态。
Synchronized在JVM里通过成对的MonitorEnter和MonitorExit指令来实现方法同步和代码块同步。
每一个Java对象自创建就带了一把看不见的锁，它叫做内部锁或者Monitor锁。
也就是通常说Synchronized的对象锁，MarkWord锁标识位为10，其中指针指向的是Monitor对象的起始地址。在HotSpot中，Monitor是由ObjectMonitor实现的。

4. 同步原理

见第三章第2节

四、加锁要考虑锁的粒度和场景问题

1. 示例

最简单的加锁方式就是在方法上添加 synchronized 关键字，但是也不能因为简单就把业务代码中的所有方法都加上synchronized，这样滥用 synchronized 是不可取的，会造成极大的性能问题。

即使确实有共享资源需要保护，也要尽可能降低锁的粒度，仅对必要的代码块甚至需要保护的资源本身加锁。

如下示例中，slow方法模拟不涉及线程安全的比较耗时的操作，正确的做法是不将slow方法同步，只同步存在线程安全问题的部分。

public class LockTest_3 {

    private static final Logger log = LoggerFactory.getLogger(LockTest_3.class);

    public static void main(String[] args) {

        LockTest_3 lockTest_3 = new LockTest_3();

        List<Integer> data = new ArrayList<>();

        Long begin = System.currentTimeMillis();
        // 多个线程执行500次
        IntStream.rangeClosed(1, 500).parallel().forEach(i -> {
            // synchronized (lockTest_3) 加在此处会大大增加执行时间
            lockTest_3.slowMethod();
            synchronized (lockTest_3) {
                data.add(i);
            }
        });
        log.info("took:{}, data.size:{}", System.currentTimeMillis() - begin, data.size());
    }

    private void slowMethod() {
        try {
            TimeUnit.MILLISECONDS.sleep(10);
        } catch (InterruptedException e) {
        }
    }
}

如果精细化考虑了锁应用范围后，性能还无法满足需求的话，就要考虑另一个维度的粒度问题了，即：区分读写场景以及资源的访问冲突，考虑使用悲观锁还是乐观锁。

对于读写比例差异明显的场景，考虑使用 ReentrantReadWriteLock 细化区分读写锁， 来提高性能。

如果 JDK 版本高于 1.8、共享资源的冲突概率也没那么大的话，考虑使用 StampedLock 的乐观读的特性，进一步提高性能。

JDK 里 ReentrantLock 和 ReentrantReadWriteLock 都提供了公平锁的版本，在没有明确需求的情况下不要轻易开启公平锁特性，在任务很轻的情况下开启公平锁可能会让性能下降上百倍。（因为设置为公平锁，会先看等待队列中有没有线程，有的话会先进行入队操作，耗费性能）

2. ReentrantReadWriteLock

读锁是共享锁，写锁是排他锁
示例如下：

/**
 * 读写锁效率测试
 */
public class LockTest_ReadWriteLock {

    private static volatile int value = 1;

    static Lock lock = new ReentrantLock();

    static ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
    static Lock readLock = readWriteLock.readLock();
    static Lock writeLock = readWriteLock.writeLock();

    // 模拟读操作
    public static void read(Lock lock) {
        try {
            lock.lock();
            TimeUnit.SECONDS.sleep(1);
            System.out.println("read over ! value: " + value);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }

    // 模拟写操作
    public static void write(Lock lock, int v) {
        try {
            lock.lock();
            TimeUnit.SECONDS.sleep(1);
            value = v;
            System.out.println("write over ! value: " + value);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }

    private static void run(Lock... lock) {
        // 起18个读线程
        IntStream.rangeClosed(1, 18)
                .forEach(i -> new Thread(() -> read(lock[0])).start());

        // 起2个写线程
        IntStream.rangeClosed(1, 2)
                .forEach(i -> new Thread(() -> write(lock[1], new Random().nextInt())).start());
    }

    public static void main(String[] args) {
        // ReentrantLock 测试
//        run(lock, lock);

        // ReentrantReadWriteLock 测试
        run(readLock, writeLock);
    }
}

五、多把锁要小心死锁问题

1. 示例

当一个业务逻辑涉及到多把锁时，容易产生死锁问题。

场景：下单操作需要锁定订单中多个商品的库存，拿到所有商品的锁 之后进行下单扣减库存操作，全部操作完成之后释放所有的锁。
现象：下单失败概率很高，失败后需要用户重新下单，极大影响了用户体验，还影响到了销量。
问题：是死锁引起的问题，背后原因是扣减库存的顺序不同，导致并发的情况下多个线程可能相互持有部分商品的锁，又等待其他线程释放另一部分商品的锁，于是出现了死锁问题。

案例代码：
定义了商品类型 Item ，每种默认1000库存，初始化10个商品对象模拟商品列表 items。

createCart 模拟购物车，随机选3个商品。

createOrder 下单逻辑为：遍历购物车中的商品依次尝试获取商品锁，最长等待3秒。获得所有商品锁后再扣减库存，否则释放获得的所有锁，返回false下单失败。

最后模拟多线程执行50次下单操作，观察日志输出

public class OrderDemo {

    private static final Logger log = LoggerFactory.getLogger(OrderDemo.class);

    private static ConcurrentHashMap<String, Item> items = new ConcurrentHashMap<>();

    static {
        // 初始化10个商品
        IntStream.range(0, 10).forEach(i -> items.put("item" + i, new Item("item" + i)));
    }

    /**
     * 商品实体
     */
    static class Item {
        // 商品名
        final String name;

        // 剩余库存
        int remaining = 1000;

        ReentrantLock lock = new ReentrantLock();

        public Item(String name) {
            this.name = name;
        }

        public String getName() {
            return name;
        }

        @Override
        public String toString() {
            return "Item{" +
                    "name='" + name + '\'' +
                    ", remaining=" + remaining +
                    '}';
        }
    }

    /**
     * 创建购物车（从初始化的10个商品中随机选3个）
     */
    private static List<Item> createCart() {
        return IntStream.rangeClosed(1, 3)
                .mapToObj(i -> "item" + ThreadLocalRandom.current().nextInt(items.size()))
                .map(name -> items.get(name)).collect(Collectors.toList());
    }

    /**
     * 创建订单
     */
    private static boolean createOrder(List<Item> order) {

        // 存放所有获得的锁
        List<ReentrantLock> locks = new ArrayList<>();

        for (Item item : order) {
            try {
                // 获得锁3秒超时
                if (item.lock.tryLock(3, TimeUnit.SECONDS)) {
                    locks.add(item.lock);
                } else {
                    locks.forEach(ReentrantLock::unlock);
                    return false;
                }
            } catch (InterruptedException e) {

            }
        }

        // 锁全部拿到之后执行扣减库存业务逻辑
        try {
            order.forEach(item -> item.remaining--);
        } finally {
            locks.forEach(ReentrantLock::unlock);
        }
        return true;
    }

    /**
     * 错误下单操作
     */
    private static void errorOperation(){
        long begin = System.currentTimeMillis();

        // 并发进行50次下单操作，统计成功次数
        long success = IntStream.rangeClosed(1, 50).parallel()
                .mapToObj(i -> {
                    List<Item> cart = createCart();
                    return createOrder(cart);
                }).filter(result -> result)
                .count();

        log.info("success:{} totalRemaining:{} took:{}ms items:{}",
                success,
                items.entrySet().stream().map(item -> item.getValue().remaining).reduce(0, Integer::sum),
                System.currentTimeMillis() - begin,
                items);
    }

    /**
     * 正确下单操作
     */
    private static void rightOperation(){
        long begin = System.currentTimeMillis();

        // 并发进行50次下单操作，统计成功次数
        long success = IntStream.rangeClosed(1, 50).parallel()
                .mapToObj(i -> {
                    List<Item> cart = createCart().stream()
                            .sorted(Comparator.comparing(Item::getName))
                            .collect(Collectors.toList());
                    return createOrder(cart);
                }).filter(result -> result)
                .count();

        log.info("success:{} totalRemaining:{} took:{}ms items:{}",
                success,
                items.entrySet().stream().map(item -> item.getValue().remaining).reduce(0, Integer::sum),
                System.currentTimeMillis() - begin,
                items);
    }

    /**
     * 模拟下单操作
     */
    public static void main(String[] args) {

//        errorOperation();

        rightOperation();

    }

}

errorOperation执行结果：

[INFO ] [main] [c.j.test.locktest.OrderDemo   ] success:35 totalRemaining:9895 took:6022ms items:{item0=Item{name='item0', remaining=988}, item2=Item{name='item2', remaining=992}, item1=Item{name='item1', remaining=992}, item8=Item{name='item8', remaining=990}, item7=Item{name='item7', remaining=989}, item9=Item{name='item9', remaining=990}, item4=Item{name='item4', remaining=988}, item3=Item{name='item3', remaining=992}, item6=Item{name='item6', remaining=991}, item5=Item{name='item5', remaining=983}}

rightOperation执行结果：

[INFO ] [main] [c.j.test.locktest.OrderDemo   ] success:50 totalRemaining:9850 took:15ms items:{item0=Item{name='item0', remaining=989}, item2=Item{name='item2', remaining=983}, item1=Item{name='item1', remaining=986}, item8=Item{name='item8', remaining=984}, item7=Item{name='item7', remaining=985}, item9=Item{name='item9', remaining=987}, item4=Item{name='item4', remaining=990}, item3=Item{name='item3', remaining=982}, item6=Item{name='item6', remaining=980}, item5=Item{name='item5', remaining=984}}

错误操作会产生死锁问题。因为多个线程如果获取商品锁的顺序不统一，可能会互相持有对方购物车中的商品锁。

死锁.png

如何避免上述的死锁问题?
解决方法很简单，为购物车中的商品排序，让所有线程都是按照一定的顺序获取锁，就能避免死锁。

2. 关于下单与减库存的顺序问题

上面提到了下单的业务，那么实际开发中，一个事务中是先进行 下单操作 还是 减库存操作 呢？

答案是应该先进行下单操作。

以MySQL数据库为例，下单就是 insert 操作，insert 插入是行级锁，支持每秒 4W 的并发。而减库存是 update 操作，命中索引时也是行级锁，但是这是个独占锁，库存可能同时会有多个线程要操作，这时所有的操作都要等待前一个释放锁后才能继续 update。

下单减库存顺序.png

问题就在这里，根据MySQL两阶段锁协议，应该把热点操作放到离 commit 近的位置，这样可以减少行锁的持有时间，处理效率更好。