物联网-异步控制多个设备

物联网-异步控制设备

背景

在这个万物互联的时代，物联网设备起到了关键性的作用，那我们怎么去联动一个个物联网设备。比如一个家庭，里面有很多的设备，比如洗衣机，加湿器、空调、除湿机、灯光等等这些智能设备。假如在一个场景中存在多个空调、多个除湿机，人们在米家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提供了很强的异步控制功能，还有很多的用法等待大家探索喔，欢迎关注和探讨。