物联网-异步控制设备
背景
在这个万物互联的时代，物联网设备起到了关键性的作用，那我们怎么去联动一个个物联网设备。比如一个家庭，里面有很多的设备，比如洗衣机，加湿器、空调、除湿机、灯光等等这些智能设备。假如在一个场景中存在多个空调、多个除湿机，人们在米家APP或者语言进行一键控制的时候，云端是如何去控制设备呢？
当然在云端会存在很多设备的控制指令，但是控制服务暴露的地址只有一个，如果我们只采用简单的遍历去控制设备，那整体的控制时间是每个设备控制的总和。如果网络稍微延迟，就会导致用户体验差。那为了避免这样，我们可以采用怎样的方式去控制设备呢？
在实际的工程中，我们采用的是一种异步的方式去控制设备，这样控制设备的时间就是单独的一个设备的花费的时长。
假如一个场景中有三个设备，每个设备控制的时间花费：X，循环遍历去控制的话花费的时间是：X+X+X，但如果用异步就是：X
如何实现异步
在云端我们采用了Java语言编写异步功能，这里使用到了CompletableFuture。CompleteableFuture是在Java8提出来的，之前为了能获取到异步执行返回的结果，都是采用Future，通过调用阻塞方法get()，要么轮训isDone是否为True，但是这两种方法会阻塞主线程，导致主线程等待。
因此为了改进Future，才引入了CompletableFuture，我们可以在主线程中定义好回调后，当异步任务执行完成或者执行出错，可以自动调用回调对象的回调方法，不存在主线程等待的现象。
因此CompletableFuture的优点有：
- 异步任务结束时，会自动回调某个对象的方法；
- 异步任务出错时，会自动回调某个对象的方法；
- 主线程设置好回调后，不再关心异步任务的执行。
举个生活中的例子：
例子1：在农村的大锅做饭的时候，需要时时刻刻，搅拌着米，最后要尝一下饭是否熟了。
例子2：我们买了一个电饭煲，设置好预约就好，饭做好了告诉我们就好。
例子1，就是之前的Future， 例子2就是CompletableFuture。
代码例子
1. 首先定义一个mock设备控制的远程服务
这是一个实际的控制服务
/**
* mock 远程调动设备控制方法
*
* @param request
* @return
*/
public static Map remoteDeviceControl(ControlRequest request) {
// todo
//1. 指令转换，转换成功后下发到电控，电控控制设备
// 省略。。。
//2. 设备响应后，上报云端解析，是否控制成功
Map result = new HashMap<>();
result.put("code", 0);
result.put("message", "success");
return result;
}
2. 异步控制一个设备的方法
/**
* 单独的异步控制
* @param request
* @return
*/
public static CompletionStage<Map> deviceControl(ControlRequest request) {
return CompletableFuture.supplyAsync(() -> {
Map result;
// 远程调用
result = remoteDeviceControl(request);
return result;
}).exceptionally(th -> {
Map result = new HashMap<>();
result.put("code", -1);
result.put("message", th.getMessage());
System.out.println("异步 异常：" + th);
return result;
});
}
在这个方法中：
supplyAsync()：使用异步执行exceptionally()：出现异常后进行回调，返回错误信息
3. 对多个设备控制进行编排，定义回调
/**
* 实现对多设备控制编排，定义回调
* @param requests
* @return
*/
public static CompletableFuture<List<Map>> actionAsync(List<ControlRequest> requests) {
CompletableFuture<List<Map>> resultList = CompletableFuture.supplyAsync(() -> requests).thenCompose(requests1 -> {
List<CompletionStage<Map>> controlRets = requests1.stream().map(request -> deviceControl(request)).collect(Collectors.toList());
CompletableFuture<Void> done = CompletableFuture.allOf(controlRets.toArray(new CompletableFuture[controlRets.size()]));
return done.thenApply(v -> controlRets.stream().map(CompletionStage::toCompletableFuture).map(CompletableFuture::join).collect(Collectors.toList()));
}).whenComplete((controlRets, th) -> {
if (th != null) {
System.out.printf("异步控制设备异常，{}", th.toString());
}
}).toCompletableFuture();
return resultList;
}
在这个方法中：
CompletableFuture.supplyAsync(() -> requests)：将入参转换成CompletableFuture格式。thenCompose()：实现两个CompletableFuture的连接和参数传递。requests1.stream().map()：结合Stream方法对设备进行控制。CompletableFuture.allOf()：编排当所有的异步控制结束才算结束。thenApply()：实现CompletableFuture类型转换，当计算完成时使用。done.thenApply(v -> controlRets.stream().map(CompletionStage::toCompletableFuture).map(CompletableFuture::join).collect(Collectors.toList()));：对done进行转换，获取控制的结果。
4. Demo测试
public static void main(String[] args) {
ControlRequest controlRequest = new ControlRequest();
Map command = new HashMap<>();
command.put("power", "on");
ControlRequest controlRequest1 = new ControlRequest();
Map command1 = new HashMap<>();
command1.put("power", "on");
controlRequest1.setCommand(command1);
List<ControlRequest> deviceList = new ArrayList<>();
deviceList.add(controlRequest);
deviceList.add(controlRequest1);
List<Map> resultList = new ArrayList<>();
resultList = actionAsync(deviceList).join();
System.out.println(resultList);
}
其中：
ControlRequest是一个控制入参实体。
public class ControlRequest {
private Map command;
public Map getCommand() {
return command;
}
public void setCommand(Map command) {
this.command = command;
}
}
- 模拟对两个设备进行开机：
command1.put("power", "on")。 resultList = actionAsync(deviceList).join()：通过join获取CompletableFuture包装的结果。
最终的运行结果：
[{code=0, message=success}, {code=0, message=success}]
结语
以上就是一个异步控制多设备的demo，CompletableFuture提供了很强的异步控制功能，还有很多的用法等待大家探索喔，欢迎关注和探讨。
