物联网-异步控制设备

背景

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