一个线程等待另一个线程的执行结果，guarded suspension
要点
- 有一个结果需要从一个线程传递到另一个线程，让他们关联同一个GuardedObject
- 如果有结果不断从一个线程到另一个线程，可以使用消息队列
- jdk中，join的实现，future的实现都是采用的保护性暂停
- 要等待另一方的结果，归类到同步模式
- 产生结果线程与消费结果线程一一对应
class GuardedObject {
private Object response;
private final Object lock = new Object();
public Object get() {
synchronized (lock) {
// 条件不满足则等待
while (response == null) {
try {
lock.wait();
} catch (InterruptedException e) {
e.printStackTrace();
} }
return response; }
}
public void complete(Object response) {
synchronized (lock) {
// 条件满足，通知等待线程
this.response = response;
lock.notifyAll();
}
}
}
public static void main(String[] args) {
GuardedObject guardedObject = new GuardedObject();
new Thread(() -> {
try {
// 子线程执行下载
List response = download();
log.debug("download complete...");
guardedObject.complete(response);
} catch (IOException e) {
e.printStackTrace();
}
}).start();
log.debug("waiting...");
// 主线程阻塞等待
Object response = guardedObject.get();
log.debug("get response: [{}] lines", ((List) response).size());
}
超时
class GuardedObjectV2 {
private Object response;
private final Object lock = new Object();
public Object get(long millis) {
synchronized (lock) {
// 1) 记录最初时间
long begin = System.currentTimeMillis();
// 2) 已经经历的时间
long timePassed = 0;
while (response == null) {
// 4) 假设 millis 是 1000，结果在 400 时唤醒了，那么还有 600 要等
long waitTime = millis - timePassed;
log.debug("waitTime: {}", waitTime);
if (waitTime <= 0) {
log.debug("break...");
break;
}
try {
lock.wait(waitTime);
} catch (InterruptedException e) {
e.printStackTrace();
}
// 3) 如果提前被唤醒，这时已经经历的时间假设为 400
timePassed = System.currentTimeMillis() - begin;
log.debug("timePassed: {}, object is null {}",
timePassed, response == null);
}
return response;
}
}
public void complete(Object response) {
synchronized (lock) {
// 条件满足，通知等待线程
this.response = response;
log.debug("notify...");
lock.notifyAll();
}
}
}
public static void main(String[] args) {
GuardedObjectV2 v2 = new GuardedObjectV2();
new Thread(() -> {
sleep(1);
v2.complete(null);
sleep(1);
v2.complete(Arrays.asList("a", "b", "c"));
}).start();
Object response = v2.get(2500);
if (response != null) {
log.debug("get response: [{}] lines", ((List) response).size());
} else {
log.debug("can't get response");
}
}
没有超时
多任务版本的解耦GuardedObject类class Mailboxes {
private static Map boxes = new Hashtable<>();
private static int id = 1;
// 产生唯一 id
private static synchronized int generateId() {
return id++;
}
public static GuardedObject getGuardedObject(int id) {
return boxes.remove(id);
}
public static GuardedObject createGuardedObject() {
GuardedObject go = new GuardedObject(generateId());
boxes.put(go.getId(), go);
return go;
}
public static Set getIds() {
return boxes.keySet();
}
}
join原理
Thread类中存在无参和有参的join()方法，其实无参的join方法体中调用的是传值为0的join()方法
具体看看有参的join方法怎么运行
- 当传入的时间小于0，非法参数，执行异常
- 当传入的时间==0，让线程一直等待，直到线程结束，通过Thread的本地方法isAlive()判断是否存活，存活就一直等待
- 当传入时间>0，记录一个开始时间，经历时间，通过传入时间-经历时间得到需要等待的时间，如果等待时间<=0，则无需等待直接break跳出循环，否则进入等待，且更新经历时间
join需要等待线程执行完毕后才会执行下一个线程，而保护性暂停执行完其他线程所要的资源后可以执行其他操作，不用等待线程全部执行完毕
join变量需为全局，而保护性暂停的资源变量可以是局部的
异步模式之生产者/消费者 要点- 与保护性暂停中的GuardObject不同，不需要产生结果和消费结果的线程一一对应
- 消费队列可以用来平衡生产和消费的线程资源
- 生产者仅负责产生结果数据，不关心数据如何被处理，消费者仅负责处理结果数据
- 消息队列有容量限制，满时不会再加入数据，空时不会再消耗数据
- jdk中各种阻塞队列采用的模式
- RabbitMq
class Message {
private int id;
private Object message;
public Message(int id, Object message) {
this.id = id;
this.message = message;
}
public int getId() {
return id;
}
public Object getMessage() {
return message;
}
}
//创建一个消息队列
class MessageQueue {
private LinkedList queue;
//队列可容纳的最大容量
private int capacity;
public MessageQueue(int capacity) {
this.capacity = capacity;
queue = new LinkedList<>();
}
//从队列中读取数据
public Message take() {
synchronized (queue) {
while (queue.isEmpty()) {
log.debug("没货了, wait");
try {
queue.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
Message message = queue.removeFirst();
queue.notifyAll();
return message;
}
}
//向队列中添加数据
public void put(Message message) {
synchronized (queue) {
while (queue.size() == capacity) {
log.debug("库存已达上限, wait");
try {
queue.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
queue.addLast(message);
queue.notifyAll();
}
}
}
应用
MessageQueue messageQueue = new MessageQueue(2);
// 4 个生产者线程, 下载任务
for (int i = 0; i < 4; i++) {
int id = i;
new Thread(() -> {
try {
log.debug("download...");
List response = Downloader.download();
log.debug("try put message({})", id);
messageQueue.put(new Message(id, response));
} catch (IOException e) {
e.printStackTrace();
}
}, "生产者" + i).start();
}
// 1 个消费者线程, 处理结果
new Thread(() -> {
while (true) {
Message message = messageQueue.take();
List response = (List) message.getMessage();
log.debug("take message({}): [{}] lines", message.getId(), response.size());
}
}, "消费者").start();
