摘要：背景通過接口實(shí)現(xiàn)調(diào)用發(fā)送數(shù)據(jù)，接口返回值為發(fā)送數(shù)據(jù)的對(duì)應(yīng)結(jié)果。接口為同步阻塞，為異步回調(diào)方式。接收數(shù)據(jù)回調(diào)接收到數(shù)據(jù)后，通過閉鎖釋放阻塞的線程，同時(shí)設(shè)置結(jié)果返回給調(diào)用者
背景
通過HTTP接口實(shí)現(xiàn)調(diào)用MQTT Client發(fā)送數(shù)據(jù)，HTTP接口返回值為MQTT Client發(fā)送數(shù)據(jù)的對(duì)應(yīng)結(jié)果。 HTTP接口為同步阻塞，MQTT Client 為異步回調(diào)方式。
如何實(shí)現(xiàn)在HTTP接口中調(diào)用MQTT Client發(fā)送數(shù)據(jù)后，能夠阻塞等待MQTT返回結(jié)果，然后將結(jié)果返回？
解決方法
CountDownLatch + Callbale+FutureTask
1.CountDownLatch作用
CountDownLatch實(shí)現(xiàn)在MQTT Client 發(fā)送數(shù)據(jù)后 到接收數(shù)據(jù)后這段時(shí)間的阻塞。
HTTP每次請(qǐng)求，新建一個(gè)CountDownLatch，然后將CountDownLatch作為值和deviceId作為KEY保存到Map中，
調(diào)用MQTT Client 發(fā)送數(shù)據(jù)后，countDownLatch.await(),進(jìn)行同步等待
在MQTT Client接收數(shù)據(jù)的回調(diào)方法中更加deviceId取出CountDwonLatch然后計(jì)數(shù)減一
2.Callbale+FutureTask作用
將調(diào)用MQTT Client發(fā)送數(shù)據(jù)的過程，封裝成Callable,投遞發(fā)送任務(wù)時(shí)，通過返回的FutureTask的get()方法，
同步阻塞，直到結(jié)果返回。
關(guān)鍵代碼
1.Map保存CountDownObj用于同步阻塞等待MQTT Client返回結(jié)果，以及將返回結(jié)果傳遞個(gè)FutureTask
private final static ConcurrentMap countDownLatchMap = new ConcurrentHashMap<>();
//線程池
private final ThreadPoolExecutor threadPoolExecutor =
new ThreadPoolExecutor(3, 5, 30, TimeUnit.SECONDS, new ArrayBlockingQueue<>(100), runnable -> {
Thread thread = new Thread(runnable, "mqtt thread");
return thread;
});
2.HTTP API 調(diào)用的發(fā)送MQTT 消息數(shù)據(jù)的接口
/**
* HTTP API 調(diào)用的發(fā)送MQTT 消息數(shù)據(jù)的接口
* 同步阻塞
*/
public Integer send(Long packageId, String deviceId) throws Exception {
......
FutureTask futureTask = sendTask(publishDto));
return futureTask.get()
}
3.投遞發(fā)送MQTT指令的task方法
/**
* 投遞MQTT發(fā)送指令任務(wù)
* 同步阻塞
*/
private FutureTask sendTask(PublishDto publishDto) throws Exception {
FutureTask futureTask = new FutureTask<>(new GetDatapointValueCallable(publishDto));
threadPoolExecutor.execute(futureTask);
//阻塞線程
return futureTask;
}
4.封裝CountDownLatch 和 Integer的對(duì)象，用于CountDownLatch阻塞控制和返回結(jié)果
/**
* 封裝CountDownLatch 和 Integer
* 用于CountDownLatch阻塞控制和返回結(jié)果
*/
private class CountDownObj {
private final CountDownLatch countDownLatch;
private volatile Integer value;
private CountDownObj(CountDownLatch countDownLatch) {
this.countDownLatch = countDownLatch;
}
public CountDownLatch getCountDownLatch() {
return countDownLatch;
}
public Integer getValue() {
return value;
}
public void setValue(Integer value) {
this.value = value;
}
}
5.具體發(fā)送MQTT數(shù)據(jù)的Callbale線程Task，會(huì)新建CountDownLatch，并通過CountDownLatch.await()方法阻塞，直到MQTT回調(diào)接收到數(shù)據(jù)或者超時(shí)。
/**
* 發(fā)送MQTT消息的任務(wù)Callable
*/
private class GetDatapointValueCallable implements Callable {
private final PublishDto publishDto;
GetDatapointValueCallable(PublishDto publishDto) {
this.publishDto = publishDto;
}
@Override
public Integer call() throws Exception {
//mqtt client 發(fā)送數(shù)據(jù),此處具體代碼省略
......
CountDownLatch countDownLatch = new CountDownLatch(1);
countDownLatchMap.putIfAbsent(publishDto.getDeviceId(), new CountDownObj(countDownLatch));
//阻塞，超時(shí)時(shí)間3s
countDownLatch.await(3, TimeUnit.SECONDS);
//返回mqtt指令對(duì)應(yīng)的結(jié)果或者null
return countDownLatchMap.remove(publishDto.getDeviceId()).getValue();
}
}
6.MQTT接收數(shù)據(jù)回調(diào)，這里通過deviceId從MAP里面取到CountDownObj，釋放閉鎖（結(jié)束callable線程的等待）和設(shè)置MQTT返回的結(jié)果（即callable中call()返回的結(jié)果，也就是FutureTask的get()方法返回的結(jié)果）。
/**
* MQTT 接收數(shù)據(jù)回調(diào)
*/
void mqttReceiveCallback(String deviceId, String datapointId, String value) {
......
//接收到數(shù)據(jù)后，通過閉鎖釋放阻塞的線程，同時(shí)設(shè)置結(jié)果返回給調(diào)用者
CountDownObj countDownObj=countDownLatchMap.get(deviceId);
if(countDownObj!=null) {
countDownObj.setValue(Integer.parseInt(value));
countDownObj.getCountDownLatch().countDown();
}
.......
}
文章版權(quán)歸作者所有，未經(jīng)允許請(qǐng)勿轉(zhuǎn)載,若此文章存在違規(guī)行為，您可以聯(lián)系管理員刪除。
轉(zhuǎn)載請(qǐng)注明本文地址：http://systransis.cn/yun/68941.html
