并发 | 扣哥哥的博客

学而不思则罔

发布日期: 2019-12-16

文章字数: 1.3k

阅读时长: 5 分

阅读次数:

并发控制

在java里,同步控制怎么做?
加锁. 专业术语叫同步器

synchronize(基于JVM底层,基于C++, 底层行为不可控)
AbstractQueueSynchronizer(不利用任何JVM内置锁), 基于java可变行为去实现同步

j.u.c.

为了实现并发访问的控制, 在Java 5版本添加了java.utils.concurrent包

非线程安全	线程安全
HashMap	ConcurrentHashMap
ArrayList	CopyOnWriteArrayList
HashSet	CopyOnWriteArraySet
priorityQueue	PriorityBlockingQueue

核心三部分

CAS
自旋
LockSupport

用CAS保证状态修改的原子性
用自旋阻塞
在自旋到一定程度时,使用LockSupport将线程阻塞,让出cpu使用权

下面做一个简单的并发情况 :

// 在一个秒杀场景下, 库存是有限的. 需要保证没有超卖的情况

public class TradeService {
    private static int stock = 5;

    public void decStockNoLock() throws InterruptedException {

        Thread current = Thread.currentThread();

        if (stock > 0){
            System.out.println(current.getName() + " : 购买成功" );
            Thread.sleep(1);
            stock = stock - 1;
            System.out.println(current.getName() + " : 库存还剩 " + stock);
            System.out.println();
        }else {
            System.out.println("购买失败, 库存不足");
        }
    }
}

public class MainTest {

    public static void main(String[] args) throws InterruptedException {
        TradeService tradeService = new TradeService();

        Thread[] threads = new Thread[10];
        for (int i = 0; i < 10; i++) {
            Thread thread = new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        tradeService.decStockNoLock();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
            thread.setName("thread - " + i);
            threads[i] = thread;
        }

        for (Thread thread : threads) {
            thread.start();
            thread.join();
        }
    }
}

输出结果 :

thread - 0 : 购买成功
thread - 0 : 库存还剩 4

thread - 4 : 购买成功
thread - 9 : 购买成功
thread - 2 : 购买成功
thread - 1 : 购买成功
thread - 3 : 购买成功
thread - 5 : 购买成功
thread - 4 : 库存还剩 3

thread - 3 : 库存还剩 2

thread - 7 : 购买成功
thread - 1 : 库存还剩 0

thread - 9 : 库存还剩 1

购买失败, 库存不足
thread - 2 : 库存还剩 -1

购买失败, 库存不足
thread - 7 : 库存还剩 -2

thread - 5 : 库存还剩 -3

可以看到, 原本库存只有5个的商品,但是卖出去了8个. 这是因为线程在争夺CPU资源的时候可能会被挂起, 这个挂起可能会出现在判断库存之后减少库存之前. 在这之间其他的线程使用未减少的库存过了if判断导致超卖的情况出现.

哪怕将库存的修饰由static改成volatile也解决不了这种问题.

这种情况下是因为获取库存数量进行判断和减少库存没有保持原子性.
要想多个操作之间保持原子性,就要将这几个操作锁在一起.即,同一时刻仅允许一个线程做该操作,将其他想要进入该锁块内的线程挂起,在操作全部做完之后再进行唤醒.

那就可以使用上面提到的同步器核心三个部分,


public class TradeService {

    private volatile int state = 0;

    private static final Unsafe UNSAFE = UnSafeInstance.reflectGetUnsafe();

    private final static long staticOffset;

    private Thread lockHolder;

    private volatile int stock = 5;

    private Queue<Thread> threadQueue = new ConcurrentLinkedDeque<>();

    static {
        try {
            // 找到"state"在类内的偏移量
            staticOffset = UNSAFE.objectFieldOffset(TradeService.class.getDeclaredField("state"));
        } catch (Exception e) {
            throw new Error(e);
        }
    }

    // 进行一次 cas原子修改
    public boolean compareAndSwapState(int expect, int update){
        //(要修改的对象, 修改的字段在类内的偏移量, 预期值, 修改值)
        return UNSAFE.compareAndSwapInt(this, staticOffset, expect, update);
    }

    public int getState(){
        return state;
    }


    public void decStockNoLock() throws InterruptedException {

        Thread current = Thread.currentThread();
        for (;;){
            int state = getState();
            if (state == 0){
                if (compareAndSwapState(0, 1)){
                    // 如果加锁成功
                    lockHolder = current;
                    break;
                }
            }

            // 如果加锁失败,把当前线程阻塞住, 释放CPU资源
            // 阻塞之前需要把线程保存起来.
            // 需要FIFO这样的队列, 避免像synchronized唤醒所有有阻塞停留在管程对象上的线程
            threadQueue.add(current);
            LockSupport.park();
        }


        // 已经加锁成功
        // todo 做一些操作

        if (stock > 0){
            System.out.println(current.getName() + " : 购买成功" );
            Thread.sleep(1);
            stock = stock - 1;
            System.out.println(current.getName() + " : 库存还剩 " + stock);
            System.out.println();
        }else {
            System.out.println("购买失败, 库存不足");
        }

        // 解锁
        for (;;){
            int state = getState();
            if (state != 0 && lockHolder == current){
                compareAndSwapState(state, 0);
                // 释放锁需要通知被阻塞的线程.
                if (threadQueue.size() > 0){
                    Thread thread = threadQueue.poll();
                    LockSupport.unpark(thread);
                }
                break;
            }
        }
    }
}

输出结果 :

thread - 1 : 购买成功
thread - 1 : 库存还剩 4

thread - 3 : 购买成功
thread - 3 : 库存还剩 3

thread - 0 : 购买成功
thread - 0 : 库存还剩 2

thread - 4 : 购买成功
thread - 4 : 库存还剩 1

thread - 8 : 购买成功
thread - 8 : 库存还剩 0

购买失败, 库存不足
购买失败, 库存不足
购买失败, 库存不足
购买失败, 库存不足
购买失败, 库存不足

这么看是解决了超卖的问题, 但是为什么要这么麻烦呢? 使用synchronized不也是一样的效果吗?