深入理解AQS-1 AQS初相识~手动实现简单公平锁

小咪咪 2023-02-20 13:52 44阅读 0赞

# 锁源码 #

package com.test.zwj;
    
    import sun.misc.Unsafe;
    
    import java.lang.reflect.Field;
    import java.util.concurrent.ConcurrentLinkedQueue;
    import java.util.concurrent.locks.LockSupport;
    
    public class MyFairLock { 
    
        //锁状态
        private int state;
    
        //持锁线程
        private Thread LockHolder;
    
        //等待队列(线程安全)
        private ConcurrentLinkedQueue<Thread> waiters = new ConcurrentLinkedQueue<>();
    
        //Unsafe
        private static final Unsafe unsafe = getUnsafe();
    
        //锁状态偏移量
        private static long stateOffset;
    
        static { 
            try { 
                stateOffset = unsafe.objectFieldOffset(MyFairLock.class.getDeclaredField("state"));
            } catch (NoSuchFieldException e) { 
               throw new Error(e);
            }
        }
    
        /** * 加锁 * 加锁成功后方法返回 */
        public void lock() { 
            //先加锁，如成功则返回，
            if (acquire()){ 
                System.out.println(Thread.currentThread().getName() + "加锁成功");
                return;
            };
            //加锁失败，则进入等待队列，等待被唤醒，
            waiters.offer(Thread.currentThread());
            //被唤醒后重新尝试加锁，因此这里使用循环,直到加锁成功，方法才能返回
            for (;;){ 
                if (acquire()){ 
                    //加锁成功，出队列
                    System.out.println(Thread.currentThread().getName() + " 加锁成功");
                    waiters.remove(Thread.currentThread());
                    return;
                }
                //睡眠一会，测试先unpark后park的情况
                /*try { TimeUnit.SECONDS.sleep(10); } catch (InterruptedException e) { e.printStackTrace(); }*/
                System.out.println(Thread.currentThread().getName() + " park");
                //阻塞当前线程，等待被唤醒
                LockSupport.park();
                //唤醒后，继续循环
            }
        }
    
        /** * 尝试加锁 * @return 是否加锁成功 */
        private boolean acquire(){ 
            if (!shouldPark() && compareAndSwapInt(0,1)){ 
                LockHolder = Thread.currentThread();
                return true;
            }
            //支持可重入
            if (Thread.currentThread() == LockHolder){ 
                state++;
                return true;
            }
            return false;
        }
    
        /** * 是否应该阻塞 * * 当已存在等待线程且自己不会首位线程，则需要等待 */
        private boolean shouldPark(){ 
            return waiters.size() != 0 && waiters.peek() != Thread.currentThread();
        }
    
        /** * 解锁 */
        public void unlock() { 
            //检验是否是持锁线程
            if (Thread.currentThread() != LockHolder){ 
                throw new RuntimeException("LockHolder is not current thread, currentThead is "
                        + Thread.currentThread().getName() + ", LockHolder thread is " + LockHolder);
            }
            //只有持锁线程能继续执行
            //清空持锁线程
            LockHolder = null;
            //恢复锁状态，必须先清空持锁线程再回复锁状态，否则可能发生：新线程设置持锁线程后，在这儿这里清空了
            state--;
            System.out.println(Thread.currentThread().getName() + " 释放锁");
            if (state == 0 ){ 
                //唤醒等待队列中第一个线程
                Thread first = waiters.peek();
                if (first != null){ 
                    //可能有人觉得这里有bug:如果线程a进入队列后即将park但是还没有park，所以唤醒将无效
                    //其实先unpark后park也没有关系，此时park()将直接返回
                    LockSupport.unpark(first);
                    System.out.println(Thread.currentThread().getName() + " 唤醒 " + first.getName());
                }
            }
        }
    
        /** * CAS更新锁状态 * @param expect 期望的当前值 * @param update 更新后的值 * @return 是否更新成功 */
        private boolean compareAndSwapInt(int expect, int update){ 
            return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
        }
    
        private static Unsafe getUnsafe(){ 
            try { 
                Field field = Unsafe.class.getDeclaredField("theUnsafe");
                field.setAccessible(true);
                return (Unsafe) field.get(null);
            } catch (Exception e) { 
                e.printStackTrace();
            }
            return null;
        }
    }

# 测试用例 #

package com.test.zwj;
    
    import java.util.concurrent.CountDownLatch;
    
    public class Main { 
        private static int counter;
        private static MyFairLock lock = new MyFairLock();
        private  static int threadNum = 10000;
        private static CountDownLatch latch = new CountDownLatch(threadNum);
        public static void main(String[] args) throws Exception { 
            for (int i = 0; i < threadNum; i++){ 
                Thread thread = new Thread(new Task());
                thread.setName("Thread-" + i);
                thread.start();
            }
            latch.await();
            System.out.println("counter is "+counter);
        }
    
        private static class Task implements Runnable{ 
    
            @Override
            public void run() { 
               fun1();
               fun2();
            }
        }
    
        private static void fun1(){ 
            lock.lock();
            try{ 
                counter++;
            }catch (Exception e){ 
                e.printStackTrace();
            }finally { 
                lock.unlock();
            }
        }
    
        private static void fun2(){ 
            lock.lock();
            try{ 
                counter++;
            }catch (Exception e){ 
                e.printStackTrace();
            }finally { 
                latch.countDown();
                lock.unlock();
            }
        }
    }