您的位置 首页 java

Java ReentranLock同步锁和Condition条件的使用

智云科技 发布于 2023年5月26日 评论关闭 阅读(80)

ReentranLock与synchronized比较

相同：

  • ReentrantLock提供了synchronized类似的功能和内存语义。

不同：

  • ReentrantLock功能性方面更全面，比如时间锁等候，可中断锁等候，锁投票等，因此更有 扩展性 。在多个条件变量和高度竞争锁的地方，用ReentrantLock更合适，ReentrantLock还提供了 Condition ，对线程的等待和唤醒等操作更加灵活，一个ReentrantLock可以有多个Condition实例，所以更有扩展性。
  • ReentrantLock 的性能比synchronized会好点。
  • ReentrantLock提供了可 轮询 的锁请求 ，他可以尝试的去取得锁，如果取得成功则继续处理，取得不成功，可以等下次运行的时候处理，所以不容易产生 死锁 ，而synchronized则一旦进入锁请求要么成功，要么一直阻塞，所以更容易产生死锁。

ReentranLock的使用

实现可轮询的锁请求

在内部锁中，死锁是致命的——唯一的恢复方法是重新启动程序，唯一的预防方法是在构建程序时不要出错。而可轮询的锁获取模式具有更完善的错误恢复机制，可以规避死锁的发生。

如果你不能获得所有需要的锁，那么使用可轮询的获取方式使你能够重新拿到控制权，它会释放你已经获得的这些锁，然后再重新尝试。可轮询的锁获取模式，由tryLock()方法实现。此方法仅在调用时锁为空闲状态才获取该锁。如果锁可用，则获取锁，并立即返回值true。如果锁不可用，则此方法将立即返回值false。此方法的典型使用语句如下：

Lock lock = …;

// 尝试获取锁请求,获取不到锁时也不会产生死锁

if (lock.tryLock()) {

try {

// manipulate protected state

} finally {

lock.unlock();

}

} else {

// perform alternative actions

}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11

实现可定时的锁请求

当使用内部锁时，一旦开始请求，锁就不能停止了，所以内部锁给实现具有时限的活动带来了风险。为了解决这一问题，可以使用 定时锁。当具有时限的活动调用了阻塞方法，定时锁能够在时间预算内设定相应的超时。如果活动在期待的时间内没能获得结果，定时锁能使程序提前返回。 可定时的锁获取模式，由tryLock(long, TimeUnit)方法实现。

实现可中断的锁获取请求

可中断的锁获取操作允许在可取消的活动中使用。lockInterruptibly()方法能够使你获得锁的时候响应中断。

ReentranLock需要注意的地方

  • lock 必须在 finally 块中释放。否则，如果受保护的代码将抛出异常，锁就有可能永远得不到释放！这一点区别看起来可能没什么，但是实际上，它极为重要。忘记在 finally 块中释放锁，可能会在程序中留下一个定时炸弹，当有一天炸弹爆炸时，您要花费很大力气才有找到源头在哪。而使用同步，JVM 将确保锁会获得自动释放
  • 当 JVM 用 synchronized 管理锁定请求和释放时，JVM 在生成线程转储时能够包括锁定信息。这些对调试非常有价值，因为它们能标识死锁或者其他异常行为的来源。 Lock 类只是普通的类，JVM 不知道具体哪个线程拥有 Lock 对象。

Condition条件的使用

条件变量很大一个程度上是为了解决Object.wait/notify/notifyAll难以使用的问题。

Condition （ 也称为条件队列 或条件变量）为线程提供了一个含义，以便在某个状态条件现在可能为 true 的另一个线程通知它之前，一直挂起该线程（即让其“等待”）。

因为访问此共享状态信息发生在不同的线程中，所以它必须受保护，因此要将某种形式的锁与该条件相关联。等待提供一个条件的主要属性是： 以原子方式 释放相关的锁，并挂起当前线程，就像 Object.wait 做的那样。

上述API说明表明条件变量需要与锁绑定，而且多个Condition需要绑定到同一锁上。 前面的Lock中提到，获取一个条件变量的方法是Lock.newCondition()。

void await() throws InterruptedException;

void awaitUninterruptibly();

long awaitNanos(long nanosTimeout) throws InterruptedException;

boolean await(long time, TimeUnit unit) throws InterruptedException;

boolean awaitUntil(Date deadline) throws InterruptedException;

void signal ();

void signalAll();

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13

以上是Condition接口定义的方法，await()对应于Object.wait，signal对应于Object.notify，signalAll对应于Object.notifyAll。特别说明的是 Condition的接口改变名称就是为了避免与Object中的wait/notify/notifyAll的语义和使用上混淆，因为Condition同样有wait/notify/notifyAll方法。

每一个Lock可以有任意数据的Condition对象，Condition是与Lock绑定的，所以就有Lock的公平性特性：如果是公平锁，线程为按照FIFO的顺序从Condition.await中释放，如果是非公平锁，那么后续的锁竞争就不保证FIFO顺序了。

使用Condition和Lock实现阻塞队列

public class BlockingQueue<T> {

private T[] mItems;

// 同步锁

private final Lock mLock = new ReentrantLock();

// 队列满时的条件

private Condition mNotFull = mLock.newCondition();

// 队列为空时条件

private Condition mNotEmpty = mLock.newCondition();

private int mCount;

private int mHeadIdx;

private int mTailIdx;

private static final int DEFAULT_QUEUE_SIZE = 10;

public BlockingQueue() {

this(DEFAULT_QUEUE_SIZE);

}

public BlockingQueue(int size) {

mItems = (T[]) new Object[size];

}

// 放入元素

public void put(T item) throws InterruptedException {

// 获取读写锁

mLock.lock();

try {

// 此处会进行阻塞,直到有元素取出

while (mCount == getCapacity()) {

// 挂起,并释放锁,条件满足时,重新获取锁

mNotFull.await();

}

mItems[mTailIdx] = item;

// 填满元素

if (++mTailIdx == getCapacity()) {

mTailIdx = 0;

}

++mCount;

// 对阻塞住的take()进行唤醒

mNotEmpty.signalAll();

} finally {

mLock.unlock();

}

}

// 取出元素

public T take() throws InterruptedException {

mLock.lock();

try {

while (mCount == 0) {

// 挂起,并释放锁,条件满足时,重新获取锁

mNotEmpty.await();

}

T item = mItems[mHeadIdx];

mItems[mHeadIdx] = null; // for GC

if (++mHeadIdx == getCapacity()) {

mHeadIdx = 0;

}

–mCount;

mNotEmpty.signalAll();

return item;

} finally {

mLock.unlock();

}

}

public int getCapacity() {

return mItems.length;

}

}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75

测试

public class Main {

private static BlockingQueue<String> sQueue;

public static void main(String[] args) {

sQueue = new BlockingQueue<>();

new Thread(() -> {

try {

// 阻塞5秒

Thread.sleep(5000);

sQueue.put(“Hello”);

} catch (InterruptedException e) {

e.printStackTrace();

}

}).start();

// 从阻塞队列中取值

try {

String item = sQueue.take();

System.out.println(item);

} catch (InterruptedException e) {

e.printStackTrace();

}

}

}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26

await() 操作

上一节中说过多次ReentrantLock是独占锁，一个线程拿到锁后如果不释放，那么另外一个线程肯定是拿不到锁，所以在lock.lock()和lock.unlock()之间可能有一次释放锁的操作（同样也必然还有一次获取锁的操作）。

我们再回头看代码，不管take()还是put()，在进入lock.lock()后唯一可能释放锁的操作就是await()了。也就是说await()操作实际上就是释放锁，然后挂起线程，一旦条件满足就被唤醒，再次获取锁！

public final void await() throws InterruptedException {

if (Thread.interrupted())

throw new InterruptedException();

Node node = addConditionWaiter();

int savedState = fullyRelease(node);

int interruptMode = 0;

while (!isOnSyncQueue(node)) {

LockSupport.park(this);

if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)

break;

}

if (acquireQueued(node, savedState) && interruptMode != THROW_IE)

interruptMode = REINTERRUPT;

if (node.nextWaiter != null)

unlinkCancelledWaiters();

if (interruptMode != 0)

reportInterruptAfterWait(interruptMode);

}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19

上面是await()的代码片段。上一节中说过,AQS在获取锁的时候需要有一个CHL的FIFO队列，所以对于一个Condition.await()而言，如果释放了锁，要想再一次获取锁那么就需要进入队列，等待被通知获取锁。完整的await()操作是安装如下步骤进行的：

  1. 将当前线程加入Condition锁队列。特别说明的是，这里不同于AQS的队列，这里进入的是Condition的FIFO队列。后面会具体谈到此结构。进行2。
  2. 释放锁。这里可以看到将锁释放了，否则别的线程就无法拿到锁而发生死锁。进行3。
  3. 自旋(while)挂起，直到被唤醒或者超时或者CACELLED等。进行4。
  4. 获取锁(acquireQueued)。并将自己从Condition的FIFO队列中释放，表明自己不再需要锁（我已经拿到锁了）。

这里再回头介绍Condition的数据结构。我们知道一个Condition可以在多个地方被await*()，那么就需要一个FIFO的结构将这些Condition串联起来，然后根据需要唤醒一个或者多个（通常是所有）。所以在Condition内部就需要一个FIFO的队列。

signal/signalAll 操作

await()清楚了，现在再来看signal/signalAll就容易多了。按照signal/signalAll的需求，就是要将Condition.await()中FIFO队列中第一个Node唤醒（或者全部Node）唤醒。尽管所有Node可能都被唤醒，但是要知道的是仍然只有一个线程能够拿到锁，其它没有拿到锁的线程仍然需要自旋等待，就上上面提到的第4步(acquireQueued)。

private void do Signal(Node first) {

do {

if ( (firstWaiter = first.nextWaiter) == null)

lastWaiter = null;

first.nextWaiter = null;

} while (!transferForSignal(first) &&

(first = firstWaiter) != null);

}

private void doSignalAll(Node first) {

lastWaiter = firstWaiter = null;

do {

Node next = first.nextWaiter;

first.nextWaiter = null;

transferForSignal(first);

first = next;

} while (first != null);

}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20

上面的代码很容易看出来，signal就是唤醒Condition队列中的第一个非CANCELLED节点线程，而signalAll就是唤醒所有非CANCELLED节点线程。当然了遇到CANCELLED线程就需要将其从FIFO队列中剔除。

标签:
打赏 点赞(2)

文章来源：智云一二三科技

文章标题：Java ReentranLock同步锁和Condition条件的使用

文章地址：https://www.zhihuclub.com/180697.shtml

关于作者: 智云科技

相关文章

热门文章

1分享新浪图床上传接口源码

点赞(878) 阅读(58,710)

2PHP简单实现路由Route功能

点赞(1.09K) 阅读(35,923)

3Tideways、xhprof 和 xhgui 打造 PHP 非侵入式监控平台

点赞(305) 阅读(4,262)

4centos系统如何查看是否安装了mysql

点赞(482) 阅读(3,734)

5curl 工具简述

点赞(451) 阅读(2,948)
网站地图