EMQ X开源版使用

Posted 2021-11-23 GW_Cheng

1 下载

进入官网

根据自己服务期的类型选择

我这里用的是centos6，用的是zip解压进行安装，选择完之后下面会有具体的安装步骤

根据步骤进行安装。

完了开放18083和1883端口

就可以访问控制台了

http://ip:18083/

默认用户名 admin，密码 public

修改密码

emqx_ctl admins passwd admin 新密码

2 设备监听

使用EMQX的WEB_HOOK进行设备监听

配置 安装目录/etc/plugins/emqx_web_hook.conf

修改web.hook.api.url ，将其改为自定义的接口
开放需要监听的事件

3 接收监听事件并记录设备状态

@Slf4j
@Api(tags = "设备相关")
@RestController
@RequestMapping("/device")
public class DeviceController {

    @PostMapping("/webHook")
    @ApiOperation("网页回调")
    public CommonResponse<String> onWebHook(@RequestBody Map<String, String> param) {
        log.info("设备回调");
        if (MapUtils.isNotEmpty(param)) {
            param.forEach((k, v) -> log.info("key:{} value:{}", k, v));
        }
        String action = param.getOrDefault("action", "");
        if (StringUtils.isNotBlank(action)) {
            MqttClient client = new MqttClient();
            client.setClientid(param.get("clientid"));
            client.setUsername(param.get("username"));
            client.setNode(param.get("node"));
            switch (action) {
                case "client_connected":// 成功连接
                    client.setState(1);
                    client.setOnlineAt(DateUtil.currentDateTime());
                    saveClientState(client);
                    break;
                case "client_disconnected": // 断开连接
                    client.setState(0);
                    client.setOfflineAt(DateUtil.currentDateTime());
                    saveClientState(client);
                    break;
                case "message_publish": // 发布消息
                    log.info("发布消息：{}", param.getOrDefault("payload", ""));
                    break;
                default:
                    break;
            }
        }
        return CommonResponse.success();
    }

	// 设备状态写入数据库的实现
    private void saveClientState(MqttClient client) {
        if (StringUtils.isBlank(client.getUsername())) {
            return;
        }
        MqttClient find = clientMapper.getByUsername(client.getUsername());
        if (Objects.isNull(find)) {
            client.setCreated(DateUtil.currentDateTime());
            clientMapper.insert(client);
        } else {
            find.setClientid(client.getClientid());
            find.setUsername(client.getUsername());
            find.setNode(client.getNode());
            find.setState(client.getState());
            if (!Objects.isNull(client.getOnlineAt())) {
                find.setOnlineAt(client.getOnlineAt());
            }
            if (!Objects.isNull(client.getOfflineAt())) {
                find.setOfflineAt(client.getOfflineAt());
            }
            clientMapper.updateById(find);
        }
    }
}

参考文献

https://docs.emqx.cn/broker/v4.3/#%E5%BC%80%E5%A7%8B%E4%BD%BF%E7%94%A8
web-hook
事件参数
https://www.freesion.com/article/9988673375/