喝了100杯醬香拿鐵，我開竅了

作者：哪吒 2023-10-09 08:53:40

使用“越細(xì)粒度的鎖越好”，真的是這樣嗎？會不會產(chǎn)生一些其它問題？通過對醬香拿鐵進(jìn)行排序，解決了死鎖問題，避免循環(huán)等待，效率也得到了提升。

大家好，我是哪吒。

上一篇提到了鎖粒度的問題，使用“越細(xì)粒度的鎖越好”，真的是這樣嗎？會不會產(chǎn)生一些其它問題？

先說結(jié)論，可能會產(chǎn)生死鎖問題。

下面還是以購買醬香拿鐵為例：

1、定義咖啡實(shí)體類Coffee

@Data
public class Coffee {
    // 醬香拿鐵
    private String name;

    // 庫存
    public Integer inventory;

    public ReentrantLock lock = new ReentrantLock();
}

2、初始化數(shù)據(jù)

private static List<Coffee> coffeeList = generateCoffee();

public static List<Coffee> generateCoffee(){
    List<Coffee> coffeeList = new ArrayList<>();
    coffeeList.add(new Coffee("醬香拿鐵1", 100));
    coffeeList.add(new Coffee("醬香拿鐵2", 100));
    coffeeList.add(new Coffee("醬香拿鐵3", 100));
    coffeeList.add(new Coffee("醬香拿鐵4", 100));
    coffeeList.add(new Coffee("醬香拿鐵5", 100));
    return coffeeList;
}

3、隨機(jī)獲取n杯咖啡

// 隨機(jī)獲取n杯咖啡
private static List<Coffee> getCoffees(int n) {
    if(n >= coffeeList.size()){
        return coffeeList;
    }

    List<Coffee> randomList = Stream.iterate(RandomUtils.nextInt(n), i -> RandomUtils.nextInt(coffeeList.size()))
            .distinct()// 去重
            .map(coffeeList::get)// 跟據(jù)上面取得的下標(biāo)獲取咖啡
            .limit(n)// 截取前面 需要隨機(jī)獲取的咖啡
            .collect(Collectors.toList());
    return randomList;
}

4、購買咖啡

private static boolean buyCoffees(List<Coffee> coffees) {
    //存放所有獲得的鎖
    List<ReentrantLock> locks = new ArrayList<>();
    for (Coffee coffee : coffees) {
        try {
            // 獲得鎖3秒超時(shí)
            if (coffee.lock.tryLock(3, TimeUnit.SECONDS)) {
                // 拿到鎖之后，扣減咖啡庫存
                locks.add(coffee.lock);
                coffeeList = coffeeList.stream().map(x -> {
                 // 購買了哪個(gè)，就減哪個(gè)
                    if (coffee.getName().equals(x.getName())) {
                        x.inventory--;
                    }
                    return x;
                }).collect(Collectors.toList());
            } else {
                locks.forEach(ReentrantLock::unlock);
                return false;
            }
        } catch (InterruptedException e) {
        }
    }
    locks.forEach(ReentrantLock::unlock);
    return true;
}

3、通過parallel并行流，購買100次醬香拿鐵，一次買2杯，統(tǒng)計(jì)成功次數(shù)

public static void main(String[] args){
    StopWatch stopWatch = new StopWatch();
    stopWatch.start();

    // 通過parallel并行流，購買100次醬香拿鐵，一次買2杯，統(tǒng)計(jì)成功次數(shù)
    long success = IntStream.rangeClosed(1, 100).parallel()
            .mapToObj(i -> {
                List<Coffee> getCoffees = getCoffees(2);
                //Collections.sort(getCoffees, Comparator.comparing(Coffee::getName));
                return buyCoffees(getCoffees);
            })
            .filter(result -> result)
            .count();

    stopWatch.stop();
    System.out.println("成功次數(shù)："+success);
    System.out.println("方法耗時(shí)："+stopWatch.getTotalTimeSeconds()+"秒");
    for (Coffee coffee : coffeeList) {
        System.out.println(coffee.getName()+"-剩余:"+coffee.getInventory()+"杯");
    }
}

耗時(shí)有點(diǎn)久啊，20多秒。

數(shù)據(jù)對不對？

醬香拿鐵1賣了53杯。
醬香拿鐵2賣了57杯。
醬香拿鐵3賣了20杯。
醬香拿鐵4賣了22杯。
醬香拿鐵5賣了19杯。
一共賣了171杯。

數(shù)量也對不上，應(yīng)該賣掉200杯才對，哪里出問題了？

4、使用visualvm測一下：

果不其然，出問題了，產(chǎn)生了死鎖。

線程 m 在等待的一個(gè)鎖被線程 n 持有，線程 n 在等待的另一把鎖被線程 m 持有。

比如美杜莎買了醬香拿鐵1和醬香拿鐵2，小醫(yī)仙買了醬香拿鐵2和醬香拿鐵1；
美杜莎先獲得了醬香拿鐵1的鎖，小醫(yī)仙獲得了醬香拿鐵2的鎖；
然后美杜莎和小醫(yī)仙接下來要分別獲取醬香拿鐵2 和醬香拿鐵1 的鎖；
這個(gè)時(shí)候鎖已經(jīng)被對方獲取了，只能相互等待一直到 3 秒超時(shí)。

5、如何解決呢？

讓大家都先拿一樣的醬香拿鐵不就好了。讓所有線程都先獲取醬香拿鐵1的鎖，然后再獲取醬香拿鐵2的鎖，這樣就不會出問題了。

也就是在隨機(jī)獲取n杯咖啡后，對其進(jìn)行排序即可。

// 通過parallel并行流，購買100次醬香拿鐵，一次買2杯，統(tǒng)計(jì)成功次數(shù)
long success = IntStream.rangeClosed(1, 100).parallel()
        .mapToObj(i -> {
            List<Coffee> getCoffees = getCoffees(2);
            // 根據(jù)咖啡名稱進(jìn)行排序
            Collections.sort(getCoffees, Comparator.comparing(Coffee::getName));
            return buyCoffees(getCoffees);
        })
        .filter(result -> result)
        .count();