• Java多线程（6）：锁与AQS（下）

    您好，我是湘王，这是我的CSDN博客，欢迎您来，欢迎您再来～

    之前说过，AQS（抽象队列同步器）是Java锁机制的底层实现。既然它这么优秀，是骡子是马，就拉出来溜溜吧。

    首先用重入锁来实现简单的累加，就像这样：

    1. /**
    2.  * 用重入锁实现累加
    3.  *
    4.  * @author 湘王
    5.  */
    6. public class MyLockTest {
    7.     private final Lock lock = new ReentrantLock();
    8.     private int value;
    9.     public int getNext() {
    10.         lock.lock();
    11.         try {
    12.             value++;
    13.         } finally {
    14.             lock.unlock();
    15.         }
    16.         return value;
    17.     }
    18.     public static void main(String[] args) {
    19.         MyLockTest myLock = new MyLockTest();
    20.         for (int i = 0; i < 5; i++) {
    21.             new Thread(new Runnable() {
    22.                 @Override
    23.                 public void run() {
    24.                     for (int i = 0; i < 5; i++) {
    25.                         System.out.println(myLock.getNext());
    26.                     }
    27.                 }
    28.             }).start();
    29.         }
    30.     }
    31. }

    运行结果显示数据有重复：

    这么简单的计算都能出现重复，这肯定是无法接受的。

    再用独占锁来试试看：

    1. public class MyExclusiveLock implements Lock {
    2. 	@Override
    3. 	public void lock() {
    4. 		
    5. 	}
    6.  
    7. 	@Override
    8. 	public void lockInterruptibly() throws InterruptedException {
    9.  
    10. 	}
    11.  
    12. 	@Override
    13. 	public boolean tryLock() {
    14. 		return false;
    15. 	}
    16.  
    17. 	@Override
    18. 	public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
    19. 		return false;
    20. 	}
    21.  
    22. 	@Override
    23. 	public void unlock() {
    24.  
    25. 	}
    26.  
    27. 	@Override
    28. 	public Condition newCondition() {
    29. 		return null;
    30. 	}
    31. }

    可以看到，实现lock接口，就需要实现若干自定义的接口。然后以内部类继承AQS的方式，实现排他锁，昨天也说过，AQS中tryAcquire()和tryRelease()是一一对应的，也就是也管获取，一个管释放，所以代码是：

    1. /**
    2.  * 内部类继承AQS的方式，实现排他锁
    3.  */
    4. private static class SyncHelper extends AbstractQueuedSynchronizer {
    5. 	private static final long serialVersionUID = -7666580981453962426L;
    6.  
    7. 	/**
    8. 	 * 第一个线程进来，拿到锁就返回true；后面的线程进来，拿不到锁就返回false
    9. 	 */
    10. 	@Override
    11. 	protected boolean tryAcquire(int arg) {
    12. 		// 获取资源状态
    13. 		int state = getState();
    14. 		if (0 == state) {// 如果没有线程拿到资源的锁
    15. 			if (compareAndSetState(0, arg)) {
    16. 				// 保存当前持有同步锁的线程
    17. 				setExclusiveOwnerThread(Thread.currentThread());
    18. 				return true;
    19. 			}
    20. 		} else if (Thread.currentThread() == getExclusiveOwnerThread()) {
    21. 			// 如果当前线程再次进来，state + 1，可重入
    22. 			// 如果这里没有这个判断，那么程序会卡死
    23. 			setState(state + arg);
    24. 			return true;
    25. 		}
    26. 		return false;
    27. 	}
    28.  
    29. 	/**
    30. 	 * 锁的获取和释放需要一一对应
    31. 	 */
    32. 	@Override
    33. 	protected boolean tryRelease(int arg) {
    34. 		// 获取资源状态
    35. 		int state = getState();
    36. 		// 返回最后一个通过setExclusiveOwnerThread()方法设置过的线程，或者null
    37. 		if (Thread.currentThread() != getExclusiveOwnerThread()) {
    38. 			throw new RuntimeException();
    39. 		}
    40. 		setState(state - arg);
    41. 		if (0 == state) {
    42. 			setExclusiveOwnerThread(null);
    43. 			return true;
    44. 		}
    45. 		return false;
    46. 	}
    47.  
    48. 	protected Condition newCondition() {
    49. 		return new ConditionObject();
    50. 	}
    51. }

    然后再用AQS实现lock接口的方法：

    1. /**
    2.  * 利用AQS实现自定义独占锁
    3.  *
    4.  * @author 湘王
    5.  */
    6. public class MyExclusiveLock implements Lock {
    7. 	private final SyncHelper synchepler = new SyncHelper();
    8.  
    9. 	@Override
    10. 	public void lock() {
    11. 		synchepler.acquire(1);
    12. 	}
    13.  
    14. 	@Override
    15. 	public void lockInterruptibly() throws InterruptedException {
    16. 		synchepler.acquireInterruptibly(1);
    17. 	}
    18.  
    19. 	@Override
    20. 	public boolean tryLock() {
    21. 		return synchepler.tryAcquire(1);
    22. 	}
    23.  
    24. 	@Override
    25. 	public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
    26. 		return synchepler.tryAcquireNanos(1, unit.toNanos(time));
    27. 	}
    28.  
    29. 	@Override
    30. 	public void unlock() {
    31. 		synchepler.release(1);
    32. 	}
    33.  
    34. 	@Override
    35. 	public Condition newCondition() {
    36. 		return synchepler.newCondition();
    37. 	}
    38.  
    39. 	/**
    40. 	 * 内部类继承AQS的方式，实现排他锁
    41. 	 */
    42. 	private static class SyncHelper extends AbstractQueuedSynchronizer {
    43. 		private static final long serialVersionUID = -7666580981453962426L;
    44. 	
    45. 		/**
    46. 		 * 第一个线程进来，拿到锁就返回true；后面的线程进来，拿不到锁就返回false
    47. 		 */
    48. 		@Override
    49. 		protected boolean tryAcquire(int arg) {
    50. 			// 获取资源状态
    51. 			int state = getState();
    52. 			if (0 == state) {// 如果没有线程拿到资源的锁
    53. 				if (compareAndSetState(0, arg)) {
    54. 					// 保存当前持有同步锁的线程
    55. 					setExclusiveOwnerThread(Thread.currentThread());
    56. 					return true;
    57. 				}
    58. 			} else if (Thread.currentThread() == getExclusiveOwnerThread()) {
    59. 				// 如果当前线程再次进来，state + 1，可重入
    60. 				// 如果这里没有这个判断，那么程序会卡死
    61. 				setState(state + arg);
    62. 				return true;
    63. 			}
    64. 			return false;
    65. 		}
    66. 	
    67. 		/**
    68. 		 * 锁的获取和释放需要一一对应
    69. 		 */
    70. 		@Override
    71. 		protected boolean tryRelease(int arg) {
    72. 			// 获取资源状态
    73. 			int state = getState();
    74. 			// 返回最后一个通过setExclusiveOwnerThread()方法设置过的线程，或者null
    75. 			if (Thread.currentThread() != getExclusiveOwnerThread()) {
    76. 				throw new RuntimeException();
    77. 			}
    78. 			setState(state - arg);
    79. 			if (0 == state) {
    80. 				setExclusiveOwnerThread(null);
    81. 				return true;
    82. 			}
    83. 			return false;
    84. 		}
    85. 	
    86. 		protected Condition newCondition() {
    87. 			return new ConditionObject();
    88. 		}
    89. 	}
    90. }

    然后再运行测试：

    1. /**
    2.  * 实现Lock接口方法并运行排他锁测试
    3.  *
    4.  * @author 湘王
    5.  */
    6. public class MyExclusiveLockTester {
    7. 	// 用自定义AQS独占锁实现
    8. 	private Lock lock = new MyExclusiveLock();
    9. 	private int value;
    10.  
    11. 	public int accmulator() {
    12. 		lock.lock();
    13. 		try {
    14. 			++value;
    15. 		} finally {
    16. 			lock.unlock();
    17. 		}
    18.  
    19. 		return value;
    20. 	}
    21.  
    22. 	public static void main(String[] args) throws InterruptedException {
    23. 		MyExclusiveLockTester test = new MyExclusiveLockTester();
    24. 		for (int i = 0; i < 5; i++) {
    25. 			new Thread(new Runnable() {
    26. 				@Override
    27. 				public void run() {
    28. 					for (int i = 0; i < 5; i++) {
    29. 						System.out.println(test.accmulator());
    30. 					}
    31. 				}
    32. 			}).start();
    33. 		}
    34. 	}
    35. }

    可以看到，结果无论怎么样都不会再重复了。

    这个只是简单的累加，接下来用AQS来实现一个实际的生活场景。比如周末带女票或男票去步行街吃饭，这时候人特别多，需要摇号，而且一次只能进去三张号（不按人头算，按叫到的号来算），该怎么实现呢？

    可以顺着这个思路：摇号机虽有很多号，但它本质上是个共享资源，很多人可以共享，但是每次共享的数量有限。这其实就是个可以指定数量的共享锁而已。

    既然有了思路，那接下来就好办了。

    1. /**
    2.  * 利用AQS实现自定义共享锁
    3.  *
    4.  * @author 湘王
    5.  */
    6. public class MyShareLock implements Lock {
    7. 	@Override
    8. 	public void lock() {
    9. 	}
    10.  
    11. 	@Override
    12. 	public void lockInterruptibly() throws InterruptedException {
    13. 	}
    14.  
    15. 	@Override
    16. 	public boolean tryLock() {
    17. 		return false;
    18. 	}
    19.  
    20. 	@Override
    21. 	public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
    22. 		return false;
    23. 	}
    24.  
    25. 	@Override
    26. 	public void unlock() {
    27. 	}
    28.  
    29. 	@Override
    30. 	public Condition newCondition() {
    31. 		return null;
    32. 	}
    33. }

    还是一样实现Lock接口，但这次是用AQS实现共享锁。

    1. /**
    2.  * 内部类继承AQS实现共享锁
    3.  *
    4.  */
    5. private static class SyncHelper extends AbstractQueuedSynchronizer {
    6. 	private static final long serialVersionUID = -7357716912664213942L;
    7.  
    8. 	/**
    9. 	 * count表示允许几个线程能同时获得锁
    10. 	 */
    11. 	public SyncHelper(int count) {
    12. 		if (count <= 0) {
    13. 			throw new IllegalArgumentException("锁资源数量必须大于0");
    14. 		}
    15. 		// 设置资源总数
    16. 		setState(count);
    17. 	}
    18.  
    19. 	/**
    20. 	 * 一次允许多少个线程进来，允许数量的线程都能拿到锁，其他的线程进入队列
    21. 	 */
    22. 	@Override
    23. 	protected int tryAcquireShared(int acquires) {
    24. 		// 自旋
    25. 		for (;;) {
    26. 			int state = getState();
    27. 			int remain = state - acquires;
    28. 			// 判断剩余锁资源是否已小于0或者CAS执行是否成功
    29. 			if (remain < 0 || compareAndSetState(state, remain)) {
    30. 				return remain;
    31. 			}
    32. 		}
    33. 	}
    34.  
    35. 	/**
    36. 	 * 锁资源的获取和释放要一一对应
    37. 	 */
    38. 	@Override
    39. 	protected boolean tryReleaseShared(int releases) {
    40. 		// 自旋
    41. 		for (;;) {
    42. 			// 获取当前state
    43. 			int current = getState();
    44. 			// 释放状态state增加releases
    45. 			int next = current + releases;
    46. 			if (next < current) {// 溢出
    47. 				throw new Error("Maximum permit count exceeded");
    48. 			}
    49. 			// 通过CAS更新state的值
    50. 			// 这里不能用setState()
    51. 			if (compareAndSetState(current, next)) {
    52. 				return true;
    53. 			}
    54. 		}
    55. 	}
    56.  
    57. 	protected Condition newCondition() {
    58. 		return new ConditionObject();
    59. 	}
    60. }

    然后再来改造之前实现的接口：

    1. /**
    2.  * 利用AQS实现自定义共享锁
    3.  *
    4.  * @author 湘王
    5.  */
    6. public class MyShareLock implements Lock {
    7. 	public static int count;
    8. 	private final SyncHelper synchepler = new SyncHelper(count);
    9.  
    10. 	@Override
    11. 	public void lock() {
    12. 		synchepler.acquireShared(1);
    13. 	}
    14.  
    15. 	@Override
    16. 	public void lockInterruptibly() throws InterruptedException {
    17. 		synchepler.acquireSharedInterruptibly(1);
    18. 	}
    19.  
    20. 	@Override
    21. 	public boolean tryLock() {
    22. 		return synchepler.tryAcquireShared(1) > 0;
    23. 	}
    24.  
    25. 	@Override
    26. 	public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
    27. 		return synchepler.tryAcquireSharedNanos(1, unit.toNanos(time));
    28. 	}
    29.  
    30. 	@Override
    31. 	public void unlock() {
    32. 		synchepler.releaseShared(1);
    33. 	}
    34.  
    35. 	@Override
    36. 	public Condition newCondition() {
    37. 		return synchepler.newCondition();
    38. 	}
    39.  
    40. 	/**
    41. 	 * 内部类继承AQS实现共享锁
    42. 	 *
    43. 	 */
    44. 	private static class SyncHelper extends AbstractQueuedSynchronizer {
    45. 		private static final long serialVersionUID = -7357716912664213942L;
    46.  
    47. 		/**
    48. 		 * count表示允许几个线程能同时获得锁
    49. 		 */
    50. 		public SyncHelper(int count) {
    51. 			if (count <= 0) {
    52. 				throw new IllegalArgumentException("锁资源数量必须大于0");
    53. 			}
    54. 			// 设置资源总数
    55. 			setState(count);
    56. 		}
    57.  
    58. 		/**
    59. 		 * 一次允许多少个线程进来，允许数量的线程都能拿到锁，其他的线程进入队列
    60. 		 */
    61. 		@Override
    62. 		protected int tryAcquireShared(int acquires) {
    63. 			// 自旋
    64. 			for (;;) {
    65. 				int state = getState();
    66. 				int remain = state - acquires;
    67. 				// 判断剩余锁资源是否已小于0或者CAS执行是否成功
    68. 				if (remain < 0 || compareAndSetState(state, remain)) {
    69. 					return remain;
    70. 				}
    71. 			}
    72. 		}
    73.  
    74. 		/**
    75. 		 * 锁资源的获取和释放要一一对应
    76. 		 */
    77. 		@Override
    78. 		protected boolean tryReleaseShared(int releases) {
    79. 			// 自旋
    80. 			for (;;) {
    81. 				// 获取当前state
    82. 				int current = getState();
    83. 				// 释放状态state增加releases
    84. 				int next = current + releases;
    85. 				if (next < current) {// 溢出
    86. 					throw new Error("Maximum permit count exceeded");
    87. 				}
    88. 				// 通过CAS更新state的值
    89. 				// 这里不能用setState()
    90. 				if (compareAndSetState(current, next)) {
    91. 					return true;
    92. 				}
    93. 			}
    94. 		}
    95.  
    96. 		protected Condition newCondition() {
    97. 			return new ConditionObject();
    98. 		}
    99. 	}
    100. }

    接下来就该测试咱们需要的效果是否能实现了：

    1. public class MyShareLockTester {
    2. 	public static void main(String[] args) throws InterruptedException {
    3. 		// 用自定义AQS共享锁实现
    4. 		// 一次允许发放三把锁
    5. 		MyShareLock.count = 3;
    6. 		final Lock lock = new MyShareLock();
    7.  
    8. 		// 模拟20个客户端访问
    9. 		for (int i = 0; i < 20; i++) {
    10. 			new Thread(new Runnable() {
    11. 				@Override
    12. 				public void run() {
    13. 					try {
    14. 						lock.lock();
    15. 						System.out.println("持有 " + Thread.currentThread().getName() + " 的客人可以进餐厅就餐");
    16. 						// 每两次叫号之间间隔一段时间，模拟真实场景
    17. 						Thread.sleep(3000);
    18. 					} catch (InterruptedException e) {
    19. 						e.printStackTrace();
    20. 					} finally {
    21. 						// 使用完成释放锁
    22. 						lock.unlock();
    23. 					}
    24. 				}
    25. 			}).start();
    26. 		}
    27. 	}
    28. }

    这里有20个号，每次只能发放3张，运行之后就可以看到确实如此。

    AQS是个很神奇也很好玩的东西，就像它的作者（也是除了高司令就是对Java影响最大的那个人，整个Java的多线程juc包代码就是他编写的）Doug Lea在AbstractQueuedSynchronizer的注释中所说：AQS只是一个框架，至于怎么玩，就是你的事了！

    感谢您的大驾光临！咨询技术、产品、运营和管理相关问题，请关注后留言。欢迎骚扰，不胜荣幸～

  • 相关阅读:
    视频抽帧转图片，opencv和ffmpeg效果测评
    杭州和智高薪招聘云计算/存储/网络工程师！
    【Pytorch】深度学习之优化器
    UG\NX二次开发 取消抑制特征 UF_MODL_unsuppress_feature
    Python点云处理（十九）点云地面点提取——CSF布料模拟算法
    七牛qshell 批量上传 mac 本地目录
    gitlub 加载慢问题处理
    docker入门配置
    【web应用系统实践】第四章作业
    股票程序化交易如何把交易分析简单化？
  • 原文地址：https://blog.csdn.net/lostrex/article/details/127604946