手写AQS非公平锁的示例分析,相信很多没有经验的人对此束手无策,为此本文总结了问题出现的原因和解决方法,通过这篇文章希望你能解决这个问题。
1. Unsafe工具类
package com.shi.flink.unsafeTest; import sun.misc.Unsafe; import java.lang.reflect.Field; /** * @author shiye * @create 2021-03-30 17:03 */ public class UnsafeUtil { public static Unsafe getInstance() { Field field = null; try { field = Unsafe.class.getDeclaredField("theUnsafe"); field.setAccessible(true); return (Unsafe) field.get(null); } catch (NoSuchFieldException e) { e.printStackTrace(); } catch (IllegalAccessException e) { e.printStackTrace(); } return null; } }
2. 手写AQS抽象类
package com.shi.flink.shilock; import com.shi.flink.unsafeTest.UnsafeUtil; import sun.misc.Unsafe; import java.util.concurrent.locks.AbstractOwnableSynchronizer; import java.util.concurrent.locks.LockSupport; /** * 自己写抽象AQS实现 * * @author shiye * @create 2021-03-30 14:10 */ public abstract class ShiAQS extends AbstractOwnableSynchronizer implements java.io.Serializable { private static final long serialVersionUID = 7373984972572414691L; /** * 头指针 */ private transient volatile Node head; /** * 尾指针 */ private transient volatile Node tail; /** * 状态值: * 0:空闲, * 1:正在有人使用 */ private volatile int state; /** * 获取当前状态 * * @return */ protected final int getState() { return state; } /** * 设置当前锁的状态 * * @param state */ public void setState(int state) { this.state = state; } /** * 使用unsafe类来初始化一些参数值 */ private static final Unsafe unsafe = UnsafeUtil.getInstance(); private static long stateOffset; private static long headOffset; private static long tailOffset; private static long waitStatusOffset; private static long nextOffset; static { try { stateOffset = unsafe.objectFieldOffset(ShiAQS.class.getDeclaredField("state")); headOffset = unsafe.objectFieldOffset(ShiAQS.class.getDeclaredField("head")); tailOffset = unsafe.objectFieldOffset(ShiAQS.class.getDeclaredField("tail")); waitStatusOffset = unsafe.objectFieldOffset(Node.class.getDeclaredField("waitStatus")); nextOffset = unsafe.objectFieldOffset(Node.class.getDeclaredField("next")); } catch (NoSuchFieldException e) { e.printStackTrace(); } } /** * 设置状态 * * @param expect * @param update * @return */ protected boolean compareAndSetState(int expect, int update) { //读取传入对象o在内存中偏移量为offset位置的值与期望值expected作比较 return unsafe.compareAndSwapInt(this, stateOffset, expect, update); } /** * 设置头指针 * * @param update * @return */ private final boolean compareAndSetHead(Node update) { return unsafe.compareAndSwapObject(this, headOffset, null, update); } /** * 如果pre节点得waitStatus值为ws, * 则把signal赋值给waitStatus * * @param pre * @param ws * @param signal * @return */ private static boolean compareAndSetWaitStatus(Node pre, int ws, int signal) { return unsafe.compareAndSwapInt(pre, waitStatusOffset, ws, signal); } /** * 设置尾指针 * * @param expect * @param update * @return */ private final boolean compareAndSetTail(Node expect, Node update) { return unsafe.compareAndSwapObject(this, tailOffset, expect, update); } /** * 设置下一个节点 * * @param node * @param expect * @param update * @return */ private static final boolean compareAndSetNext(Node node, Node expect, Node update) { return unsafe.compareAndSwapObject(node, nextOffset, expect, update); } /** * 解锁方法 * * @param arg */ protected void release(int arg) throws Exception { //尝试去释放占用锁得线程 boolean tryRelease = tryRelease(arg); if (tryRelease) { Node h = head; if (h != null && h.waitStatus != 0) { unparkSuccessor(h); } } } /** * 尝试去释放占用锁得线程 * * @param arg * @return * @throws Exception */ protected boolean tryRelease(int arg) throws Exception { if (Thread.currentThread() != getExclusiveOwnerThread()) { //如果当前线程不是占用锁得线程就抛出异常 throw new Exception("解锁失败,当前线程不是占用锁得线程无法解锁"); } else { setExclusiveOwnerThread(null); this.setState(0); return true; } } /** * 打断某个线程 * * @return */ protected boolean interruptThread(Thread thread) throws Exception { Thread ownerThread = getExclusiveOwnerThread(); if (ownerThread == thread) { //如果是正在运行得线程 compareAndSetState(1, 0); setExclusiveOwnerThread(null); } else if (head != null) { //再对类中查找当前线程,并且取消排队 for (Node next1 = head.next; next1 != null; next1 = next1.next) { if (next1.thread == thread) { compareAndSetWaitStatus(next1, next1.waitStatus, 1); } } } //解锁 thread.interrupt(); System.out.println(thread.getName() + " 已经中断了 ====> "); unparkSuccessor(head); System.out.println(thread.getName() + " 已经结束了 ====> "); return false; } /** * 自定义一个内部类Node节点 */ static final class Node { //共享模式标记 static final Node shared = new Node(); //独占锁标记 static final Node excusive = null; //waitStatus值,指示线程已取消 static final int cancelled = 1; //waitStatus值,用于指示后续线程需要解除等待状态 static final int signal = -1; //waitStatus值,指示线程正在等待条件 static final int condition = -2; //waitStatus值,指示下一个acquireShared应该 无条件传播 static final int propagate = -3; //锁的等待状态 volatile int waitStatus; //前指针 volatile Node prev; //后指针 volatile Node next; //线程 volatile Thread thread; Node nextWaiter; //是否是共享锁 final boolean isShared() { return nextWaiter == shared; } //无参构造 public Node() { } public Node(Node nextWaiter, Thread thread) { this.nextWaiter = nextWaiter; this.thread = thread; } /** * 获取前节点 * * @return */ public Node getPrev() { Node p = prev; if (p == null) { throw new NullPointerException("前节点不能为空"); } return p; } } /** * 获得 * * @param arg */ public void acquire(int arg) { //1.尝试去排队 boolean tryAcquire = tryAcquire(arg); if (!tryAcquire) { //2.如果抢占锁失败,就去排队 Node node = addWaiter(Node.excusive); //3.对已经再队列中的节点,进行休眠等侯 acquireQueued(node, arg); } } /** * 先尝试去排队 * 1.先获取锁得状态,如果状态为0,就尝试去占用一次锁 * 否则返回占用失败 * * @param arg * @return true:表示抢占锁成功 * false:表示抢占所失败 */ public final boolean tryAcquire(int arg) { Thread current = Thread.currentThread(); int state = getState(); if (state == 0) { //如果空闲了,就尝试去占用一次锁 if (compareAndSetState(0, arg)) { //抢占成功就返回true,并设置线程 setExclusiveOwnerThread(current); return true; } } else if (getExclusiveOwnerThread() == current) { //如果当前当前线程多次抢占锁,就将状态+arg int nextState = state + arg; if (nextState < 0) { throw new Error("超过最大锁计数"); } setState(nextState); return true; } return false; } /** * 添加等待队列 * * @param mode */ public Node addWaiter(Node mode) { Node node = new Node(mode, Thread.currentThread()); Node temp = tail; if (temp == null) { //入队 enQueue(node); return node; } else { //如果队列中不为空,就把当前节点添加到尾节点中 node.prev = temp; if (compareAndSetTail(temp, node)) { temp.next = node; return node; } } return node; } /** * 入队 * 把node节点添加到队列中, * 如果队列为null就初始化一个队列并且把node节点添加到尾节点中 * * @param node 返回当前节点 */ public Node enQueue(Node node) { while (true) { Node temp = tail; if (temp == null) { //创建一个头指针 compareAndSetHead(new Node()); //让尾指针也指向头指针(空节点) tail = head; } else { node.prev = temp; if (compareAndSetTail(temp, node)) { temp.next = node; return node; } } } } /** * @param node 当前正在侯队中得节点 * @param arg * @return */ protected boolean acquireQueued(Node node, int arg) { boolean failed = true; try { //是否被打断,默认false boolean interrupted = false; while (true) { final Node p = node.getPrev(); if (p == head && tryAcquire(arg)) { //如果是他的头节点是head,并且尝试抢占锁成功就出队,让当前线程运行 setHead(node); p.next = null;//利于gc回收 failed = false; return interrupted; } if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) { //pre节点得waitStatus 设置成-1,并且让当前线程阻塞,打断当前线程 interrupted = true; } } } finally { if (failed) cancelAcquire(node); } } protected final void cancelAcquire(Node node) { if (node == null) return; node.thread = null; Node pre = node.prev; while (pre.waitStatus > 0) { node.prev = pre = pre.prev; } Node predNext = pre.next; node.waitStatus = Node.cancelled; if (node == tail && compareAndSetTail(node, pre)) { compareAndSetNext(pre, predNext, null); } else { // If successor needs signal, try to set pred's next-link // so it will get one. Otherwise wake it up to propagate. int ws; if (pre != head && ((ws = pre.waitStatus) == Node.signal || (ws <= 0 && compareAndSetWaitStatus(pre, ws, Node.signal))) && pre.thread != null) { Node next = node.next; if (next != null && next.waitStatus <= 0) compareAndSetNext(pre, predNext, next); } else { unparkSuccessor(node); } node.next = node; // help GC } } /** * 解锁必须成功 * * @param node */ protected final void unparkSuccessor(Node node) { int ws = node.waitStatus; if (ws < 0) { compareAndSetWaitStatus(node, ws, 0); } /** * AQS源码是这样实现得 * 如果当前节点不为空,并且用户取消了,就从尾节点往前遍历一个,直到找到最前面得一个节点,解锁当前线程 */ Node next1 = node.next; // if (next1 == null || next1.waitStatus > 0) { // next1 = null; // for (Node t = tail; t != null && t != node; t = t.prev) { // if (t.waitStatus <= 0) { // next1 = t; // } // } // } /** * 我自己实现,从前往后找 */ if (next1 != null && next1.waitStatus > 0) { for (next1 = next1.next; next1 != null; next1 = next1.next) { if (next1.waitStatus <= 0) { break; } } } if (next1 != null) { //唤醒下一个线程 System.out.println(next1.thread.getName() + " 开始唤醒了 ====> "); LockSupport.unpark(next1.thread); System.out.println(next1.thread.getName() + " 已经唤醒了 ====> "); } } /** * 将pre节点得waitStatus 设置成-1 * * @param pre * @param node * @return */ protected static boolean shouldParkAfterFailedAcquire(Node pre, Node node) { //获取node节点得前一个节点得状态 int ws = pre.waitStatus; //如果是-1 就返回true if (ws == Node.signal) { return true; } if (ws > 0) { do { pre = pre.prev; node.prev = pre; } while (pre.waitStatus > 0); pre.next = node; } else { //设置成-1 boolean flag = compareAndSetWaitStatus(pre, ws, Node.signal); // System.out.println("设置成-1是否成功:" + flag); } return false; } /** * 阻塞当前线程,并且返回当前线程得打断状态 * * @return true: 打断线程成功 */ protected final boolean parkAndCheckInterrupt() { //打断线程,让线程阻塞 LockSupport.park(this); return Thread.interrupted(); } public Node getHead() { return head; } public void setHead(Node head) { this.head = head; } }
3.非公平所实现
package com.shi.flink.shilock; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; /** * 自定义非公平锁 * @author shiye * @create 2021-03-30 11:02 */ public class ShiNonfairLock extends ShiAQS implements Lock, java.io.Serializable { private static final long serialVersionUID = 7373984872572414699L; @Override public void lock() { if(compareAndSetState(0, 1)){ //如果抢到了锁,就把当前线程设置进去 setExclusiveOwnerThread(Thread.currentThread()); }else{ // 否则就去排队 acquire(1); } } @Override public void lockInterruptibly() throws InterruptedException { } @Override public boolean tryLock() { return false; } @Override public boolean tryLock(long time, TimeUnit unit) throws InterruptedException { return false; } @Override public void unlock() { try { super.release(1); } catch (Exception e) { e.printStackTrace(); } } @Override public Condition newCondition() { return null; } /** * 打断某个线程(自己瞎写的,有bug) * @param thread * @return */ public boolean interruptThread(Thread thread) throws Exception { return super.interruptThread(thread); } }
4.测试
package com.shi.flink.shilock; import java.util.concurrent.TimeUnit; /** * @author shiye * @create 2021-03-31 17:09 */ public class MyLockTest { public static void main(String[] args) throws Exception { ShiNonfairLock lock = new ShiNonfairLock(); new Thread(() -> { try { System.out.println("A 线程进入到...加锁过程"); lock.lock(); System.out.println("A 已经抢占到锁...休眠10s后运行......"); TimeUnit.SECONDS.sleep(10); } catch (Exception e) { e.printStackTrace(); } finally { System.out.println("A线程运行完成,开始解锁...."); lock.unlock(); } }, "A").start(); TimeUnit.SECONDS.sleep(1); Thread B = new Thread(() -> { try { System.out.println("B 线程进入到...加锁过程"); lock.lock(); System.out.println("B 已经抢占到锁...休眠10s后运行......"); TimeUnit.SECONDS.sleep(10); } catch (Exception e) { e.printStackTrace(); } finally { System.out.println("B线程运行完成,开始解锁...."); lock.unlock(); } }, "B"); B.start(); TimeUnit.SECONDS.sleep(1); new Thread(() -> { try { System.out.println("C 线程进入到...加锁过程"); lock.lock(); System.out.println("C 已经抢占到锁...休眠10s后运行......"); TimeUnit.SECONDS.sleep(10); } catch (Exception e) { e.printStackTrace(); } finally { System.out.println("C线程运行完成,开始解锁...."); lock.unlock(); } }, "C").start(); TimeUnit.SECONDS.sleep(1); new Thread(() -> { try { System.out.println("D 线程进入到...加锁过程"); lock.lock(); System.out.println("D 已经抢占到锁...休眠10s后运行......"); TimeUnit.SECONDS.sleep(10); } catch (Exception e) { e.printStackTrace(); } finally { System.out.println("D线程运行完成,开始解锁...."); lock.unlock(); } }, "D").start(); // TimeUnit.SECONDS.sleep(1); // System.out.println("强制让 " + B.getName() + " 线程中断..."); // lock.interruptThread(B); } }
看完上述内容,你们掌握手写AQS非公平锁的示例分析的方法了吗?如果还想学到更多技能或想了解更多相关内容,欢迎关注亿速云行业资讯频道,感谢各位的阅读!
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。