使用ReentrantLock可以替代内置锁,当使用内置锁的时候,我们可以使用wait() nitify()和notifyAll()来控制线程之间的协作,那么,当我们使用ReentrantLock的时候,我们怎么来处理线程之间的写作呢?
JDK5.0为我们提供了Condition对象来替代内置锁的 wait(),notify()和notifyAll()方法
内置锁的话,就只能有一个等待队列,所有的在某个对象上执行wait()方法的线程都会被加入到该对象的等待队列中去(线程会被挂起),需要其他的线程在同一个对象上调用notify()或者是notifyAll()方法来唤醒等待队列中的线程
而使用Condition的话,可以使用不同的等待队列,只需要使用lock.newCondition()即可定义一个Condition对象,每一个Condition对象上都会有一个等待队列(底层使用AQS),调用某个Condition对象的await()方法,就可以把当前线程加入到这个Condition对象的等待队列上
其他的线程调用同一个Condition对象的sinal()或者是signalAll()方法则会唤醒等待队列上的线程,使其能够继续执行
我们以一个现实中的例子来说明若何使用ReentrantLock和Condition如何替代synchronized和wait(),notify(),notifyAll():
我们模拟两个线程,一个线程执行登录操作,该登录操作会阻塞,然后等待另外一个线程将其唤醒(类似扫描登录的场景,页面会阻塞,等待扫码和确认,然后页面才会跳转)
首先是使用内置锁的例子:
package com.jiaoyiping.baseproject.condition;
import java.io.Serializable;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
/**
* Created with Intellij IDEA
*
* @author: jiaoyiping
* Mail: [email protected]
* Date: 2019/04/12
* Time: 15:29
* To change this template use File | Settings | Editor | File and Code Templates
*/
//使用内置锁来实现的等待/通知模型
public class LoginServiceUseInnerLock {
private ConcurrentHashMap<String, Result> loginMap = new ConcurrentHashMap<>();
public static void main(String[] args) throws InterruptedException {
CountDownLatch countDownLatch = new CountDownLatch(2);
LoginServiceUseInnerLock loginService = new LoginServiceUseInnerLock();
String uuid = UUID.randomUUID().toString();
System.out.println("[" + Thread.currentThread().getName() + "] 使用的UUID是: " + uuid);
new Thread(() -> {
loginService.login(uuid, 20_000);
countDownLatch.countDown();
}, "登录线程").start();
Thread.sleep(2_000);
new Thread(() -> {
loginService.confirm(uuid);
countDownLatch.countDown();
}, "确认线程").start();
countDownLatch.await();
System.out.println("[" + Thread.currentThread().getName() + "] 两个线程都执行完毕了");
}
public void login(String code, int timeout) {
Result result = new Result();
result.setMessage("超时");
loginMap.put(code, result);
synchronized (result) {
try {
//超时的话,会自动返回,程序继续
System.out.println("[" + Thread.currentThread().getName() + "] 登录线程挂起");
result.wait(timeout);
System.out.println("[" + Thread.currentThread().getName() + "] 登录线程继续执行,得到的结果是:" + result.getMessage());
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
loginMap.remove(code);
}
}
}
public void confirm(String code) {
assert code != null;
Result result = loginMap.get(code);
if (result == null) {
System.out.println("[" + Thread.currentThread().getName() + "] 请求不存在或者已经过期");
return;
}
result.setMessage("成功");
synchronized (result) {
//唤醒等待队列上的线程
System.out.println("[" + Thread.currentThread().getName() + "] 确认线程开始唤醒阻塞的线程");
result.notify();
}
}
class Result implements Serializable {
private static final long serialVersionUID = -4279280559711939661L;
String message;
public String getMessage() {
return message;
}
public void setMessage(String message) {
this.message = message;
}
public Result() {
}
public Result(String message) {
this.message = message;
}
}使用内置锁的时候,我们把random生成的key和一个自己定义的Result对象放置到ConcurrentHashMap中去,登录线程调用 Result对象的wait(timeout) 方法将当前线程挂起,并加入到Result对象的等待队列上去
确认线程根据key值,找到对应的Result对象,设置好message,然后调用Result对象的notify()方法唤醒等待队列上的线程,登录线程得以继续执行
那我们如何使用ReentrantLock和Condition来重写这个例子:
package com.jiaoyiping.baseproject.condition;
import java.io.Serializable;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
/**
* Created with Intellij IDEA
*
* @author: jiaoyiping
* Mail: [email protected]
* Date: 2019/04/12
* Time: 14:56
* To change this template use File | Settings | Editor | File and Code Templates
*/
//使用ReentrantLock和Condition来实现的等待/通知模型
public class LoginServiceUseCondition {
private ReentrantLock lock = new ReentrantLock();
ConcurrentHashMap<String, Result> conditions = new ConcurrentHashMap<>();
public static void main(String[] args) throws InterruptedException {
LoginServiceUseCondition loginService = new LoginServiceUseCondition();
String uuid = UUID.randomUUID().toString();
System.out.println("[" + Thread.currentThread().getName() + "] 使用的UUID是:" + uuid);
CountDownLatch countDownLatch = new CountDownLatch(2);
new Thread(() -> {
loginService.login(uuid, 30_000);
countDownLatch.countDown();
}, "登录线程").start();
Thread.sleep(5_000);
new Thread(() -> {
try {
Thread.sleep(3_000);
} catch (InterruptedException e) {
e.printStackTrace();
}
loginService.confirm(uuid);
countDownLatch.countDown();
}, "确认线程").start();
countDownLatch.await();
System.out.println("[" + Thread.currentThread().getName() + "] 两个线程都执行完毕了,退出");
}
/**
* 过了超时时间之后,锁会自动释放
*
* @param code
* @param timeout
*/
public void login(String code, int timeout) {
assert code != null;
try {
lock.tryLock(timeout, TimeUnit.MILLISECONDS);
Condition condition = lock.newCondition();
Result result = new Result("超时", condition);
conditions.put(code, result);
System.out.println("[" + Thread.currentThread().getName() + "] login()的请求开始阻塞");
condition.await();
System.out.println("[" + Thread.currentThread().getName() + "] 结束等待,继续执行,拿到的结果是" + result.getMessage());
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
//确认线程(拿这个UUID,去找到对应的Condition,唤醒上边的等待队列,并把Condition对象移除掉)
public void confirm(String code) {
assert code != null;
Result result = conditions.get(code);
Condition condition = result.getCondition();
if (condition != null) {
try {
System.out.println("[" + Thread.currentThread().getName() + "] 找到对应的Condition对象,将其等待队列中的线程唤醒");
lock.lock();
result.setMessage("成功");
condition.signal();
conditions.remove(code);
} finally {
lock.unlock();
}
}
}
class Result implements Serializable {
String message;
final Condition condition;
public Result(String message, Condition condition) {
this.message = message;
this.condition = condition;
}
public Result(Condition condition) {
this.condition = condition;
}
public String getMessage() {
return message;
}
public void setMessage(String message) {
this.message = message;
}
public Condition getCondition() {
return condition;
}
}
} 上边的例子说明了怎么使用ReentrantLock和Condition来代替内置锁和wait(),notify(),notifyAll()
下边的一个来自jdk中的例子,演示了如何使用同一个ReentrantLock上的多个等待队列的情况
来自JDK文档中的示例(我稍加改造,加上了main方法和一些日志):
package com.jiaoyiping.baseproject.condition;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.IntStream;
/**
* Created with Intellij IDEA
*
* @author: jiaoyiping
* Mail: [email protected]
* Date: 2019/04/12
* Time: 21:17
* To change this template use File | Settings | Editor | File and Code Templates
*/
public class BoundedBuffer {
final Lock lock = new ReentrantLock();
final Condition notFull = lock.newCondition();
final Condition notEmpty = lock.newCondition();
final Object[] items = new Object[10];
int putptr, takeptr, count;
public static void main(String[] args) throws InterruptedException {
BoundedBuffer boundedBuffer = new BoundedBuffer();
CountDownLatch countDownLatch = new CountDownLatch(40);
//分别启动20个put线程和20个take线程
IntStream.rangeClosed(1, 20).forEach(i -> {
new Thread(() -> {
try {
boundedBuffer.put(new Object());
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
countDownLatch.countDown();
}
}, "put线程 - " + i).start();
});
IntStream.rangeClosed(1, 20).forEach(i -> {
new Thread(() -> {
try {
boundedBuffer.take();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
countDownLatch.countDown();
}
}, "take线程-" + i).start();
});
countDownLatch.await();
System.out.println("[" + Thread.currentThread().getName() + "] 所有线程都执行完毕,退出");
}
public void put(Object x) throws InterruptedException {
lock.lock();
try {
//put的线程,当队列满的时候挂起
while (count == items.length) {
System.out.println("[" + Thread.currentThread().getName() + "] 线程挂起");
notFull.await();
}
Thread.sleep(1_000);
items[putptr] = x;
if (++putptr == items.length) {
putptr = 0;
}
++count;
System.out.println("[" + Thread.currentThread().getName() + "] 执行完毕写操作,唤醒take线程");
notEmpty.signal();
} finally {
lock.unlock();
}
}
public Object take() throws InterruptedException {
lock.lock();
try {
//take的线程,当队列为空的时候,挂起
while (count == 0) {
System.out.println("[" + Thread.currentThread().getName() + "] 线程挂起");
notEmpty.await();
}
Thread.sleep(1_000);
Object x = items[takeptr];
if (++takeptr == items.length) {
takeptr = 0;
}
--count;
System.out.println("[" + Thread.currentThread().getName() + "] 执行完毕读操作,唤醒put线程");
notFull.signal();
return x;
} finally {
lock.unlock();
}
}
} 我们看到,以上代码中,使用到了两个Condition:notFull和notEmpty,都是通过lock对象的newCondition()方法得来的
items被放满之后,put的线程会在notFull的等待队列上进行等待(执行了notFull.await()方法) put线程执行完操作之后,会调用 notEmpty.signal()来试图唤醒在notEmpty上等待的线程(也就是给take线程发了一个信号,告诉它,items不是空的了,你可以过来take了)
当item空了之后,take线程会在notEmpty的等待队列上进行等待(执行了notEmpty的await()方法) 当take线程执行完操作之后,会调用notFull.signal()来唤醒在notFull上等待的线程(也就是给put线程发一个信号,告诉它,items不满了,你可以进行put操作了)
和内置方法类似,在调用await(),signal(),signalAll()等方法的时候,也必须要获得锁,也就是必须在 lock.lock()和lock.unlock()代码块儿之间才能调用这些方法,否则就会抛出IllegalMonitorStateException
原文地址：https://www.cnblogs.com/jiaoyiping/p/10699273.html
时间： 2024-12-31 03:55:58