摘要：背景通过接口实现调用发送数据，接口返回值为发送数据的对应结果。接口为同步阻塞，为异步回调方式。接收数据回调接收到数据后，通过闭锁释放阻塞的线程，同时设置结果返回给调用者
背景
通过HTTP接口实现调用MQTT Client发送数据，HTTP接口返回值为MQTT Client发送数据的对应结果。 HTTP接口为同步阻塞，MQTT Client 为异步回调方式。
如何实现在HTTP接口中调用MQTT Client发送数据后，能够阻塞等待MQTT返回结果，然后将结果返回？
解决方法
CountDownLatch + Callbale+FutureTask
1.CountDownLatch作用
CountDownLatch实现在MQTT Client 发送数据后 到接收数据后这段时间的阻塞。
HTTP每次请求，新建一个CountDownLatch，然后将CountDownLatch作为值和deviceId作为KEY保存到Map中，
调用MQTT Client 发送数据后，countDownLatch.await(),进行同步等待
在MQTT Client接收数据的回调方法中更加deviceId取出CountDwonLatch然后计数减一
2.Callbale+FutureTask作用
将调用MQTT Client发送数据的过程，封装成Callable,投递发送任务时，通过返回的FutureTask的get()方法，
同步阻塞，直到结果返回。
关键代码
1.Map保存CountDownObj用于同步阻塞等待MQTT Client返回结果，以及将返回结果传递个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 调用的发送MQTT 消息数据的接口
/**
* HTTP API 调用的发送MQTT 消息数据的接口
* 同步阻塞
*/
public Integer send(Long packageId, String deviceId) throws Exception {
......
FutureTask futureTask = sendTask(publishDto));
return futureTask.get()
}
3.投递发送MQTT指令的task方法
/**
* 投递MQTT发送指令任务
* 同步阻塞
*/
private FutureTask sendTask(PublishDto publishDto) throws Exception {
FutureTask futureTask = new FutureTask<>(new GetDatapointValueCallable(publishDto));
threadPoolExecutor.execute(futureTask);
//阻塞线程
return futureTask;
}
4.封装CountDownLatch 和 Integer的对象，用于CountDownLatch阻塞控制和返回结果
/**
* 封装CountDownLatch 和 Integer
* 用于CountDownLatch阻塞控制和返回结果
*/
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.具体发送MQTT数据的Callbale线程Task，会新建CountDownLatch，并通过CountDownLatch.await()方法阻塞，直到MQTT回调接收到数据或者超时。
/**
* 发送MQTT消息的任务Callable
*/
private class GetDatapointValueCallable implements Callable {
private final PublishDto publishDto;
GetDatapointValueCallable(PublishDto publishDto) {
this.publishDto = publishDto;
}
@Override
public Integer call() throws Exception {
//mqtt client 发送数据,此处具体代码省略
......
CountDownLatch countDownLatch = new CountDownLatch(1);
countDownLatchMap.putIfAbsent(publishDto.getDeviceId(), new CountDownObj(countDownLatch));
//阻塞，超时时间3s
countDownLatch.await(3, TimeUnit.SECONDS);
//返回mqtt指令对应的结果或者null
return countDownLatchMap.remove(publishDto.getDeviceId()).getValue();
}
}
6.MQTT接收数据回调，这里通过deviceId从MAP里面取到CountDownObj，释放闭锁（结束callable线程的等待）和设置MQTT返回的结果（即callable中call()返回的结果，也就是FutureTask的get()方法返回的结果）。
/**
* MQTT 接收数据回调
*/
void mqttReceiveCallback(String deviceId, String datapointId, String value) {
......
//接收到数据后，通过闭锁释放阻塞的线程，同时设置结果返回给调用者
CountDownObj countDownObj=countDownLatchMap.get(deviceId);
if(countDownObj!=null) {
countDownObj.setValue(Integer.parseInt(value));
countDownObj.getCountDownLatch().countDown();
}
.......
}
文章版权归作者所有，未经允许请勿转载,若此文章存在违规行为，您可以联系管理员删除。
转载请注明本文地址：https://www.ucloud.cn/yun/68941.html
