前面两篇是搭建环境，从这一篇起就要真正开始我们的项目实现了，来，再一起喊下我们的口号：
人至践则无敌。
本篇文章来重点体验下 EMQX 为我们提供的 WebHook 功能。我们用到了以下技术点：
- EMQX 的 WebHook 消息监听与转发功能；
- 使用 Node.js 模拟了一个 MQTT 客户端；
- 使用 SpringBoot 搭建后端服务回调接口；
让我们看看设备消息现在到哪里了（红色背景部分）。
配置WebHook# 养成好习惯，修改配置文件前先备份
[root@iot1 emqx]# cp ./etc/plugins/emqx_web_hook.conf ./etc/plugins/emqx_web_hook.conf.bk
# 修改WebHook配置：
# 修改URL配置为自己的后端回调接口地址，注意这里的IP地址为后端服务所在的宿主机IP
# 放行这一行：web.hook.rule.message.publish.1
[root@iot1 emqx]# vi ./etc/plugins/emqx_web_hook.conf
web.hook.url = http://10.1.17.109:8000/emqx/webhook
web.hook.rule.message.publish.1 = {"action": "on_message_publish"}
# 重新加载WebHook配置
[root@iot1 emqx]# ./bin/emqx_ctl plugins reload emqx_web_hook
Plugin emqx_web_hook reloaded successfully.
之后可在 EMXQ 的 Dashboard 的 Plugins 菜单查看已开启的插件列表。
实时数据源数据源这里使用 Node.js 模拟了一个 MQTT 客户端，当然也可以使用很多 MQTT 客户端工具： HiveMQ , MQTTfx , MQTTX 等。这里简单说明下通过 MQTT 发送的数据内容，是后续将要存到 TDengine 中的数据样例，包括这样几个属性：时间戳、电压、电流、温度、设备序列号、城市、设备组编号。关于数据的建模、应用场景会在后续数据存储的章节中进行详细说明。
let mqtt = require("mqtt");
let client = mqtt.connect("mqtt://iot1", {
clientId: "mqtt_publisher_id_1",
clean: false
});
client.on("connect", function(connack) {
console.log(`return code: ${connack.returnCode}, sessionPresent: ${connack.sessionPresent}`);
if (connack.returnCode == 0) {
let data = [{
"ts": "2021-01-09 14:05:11.138",
"voltage": 225,
"currente": 9.475,
"temperature": 13.52135,
"sn": 1100,
"city": "西安",
"groupid": 1
},
{
"ts": "2021-01-09 14:05:41.138",
"voltage": 208,
"currente": 6.44464,
"temperature": 0.99233,
"sn": 1101,
"city": "西安",
"groupid": 1
},
{
"ts": "2021-01-09 14:06:11.138",
"voltage": 237,
"currente": 8.77516,
"temperature": 1.33402,
"sn": 1102,
"city": "西安",
"groupid": 1
}
];
setInterval(function() {
let index = Math.floor((Math.random() * data.length));
client.publish("device/sn", JSON.stringify(data[index]), {
qos: 1
}, function(err) {
if (err == undefined) {
console.log("Publish finished", index);
} else {
console.log("Publish failed", err);
client.end()
}
});
}, 3000);
} else {
console.log(`Connection failed: ${connack.returnCode}`)
client.end()
}
})
这里用到了 Node.js 模拟的一个 MQTT 客户端：与 MQTT Broker 即 EMQX 建立连接后，每隔3秒向主题 device/sn 随机发送一条设备数据。 MQTT Broker 收到消息后通过 EMQX 的 WebHook 插件回调后端服务接口。
搭建后端工程创建一个 SpringBoot 的空项目（这个项目也将作为后续文章中项目开发的基础），只选择了 web 与 lombok 依赖。
pom依赖配置端口org.springframework.boot spring-boot-starter-web org.projectlombok lombok true org.springframework.boot spring-boot-starter-test test
server: port: 8000回调接口
我们在后端服务中实现了一个 empx/webhook 的 RESTful 接口，通过 @RequestBody 注解以及 HashMap 类型的方式接收传进来的参数，并对收到的参数打印、返回。
切记：这里有个TODO，后续实现将借助 Spring 的事件及监听机制，实现消息入库与 WebSocket 实时推送数据至前端页面。
package com.heartsuit.emqx.controller;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.HashMap;
@RestController
@Slf4j
@RequestMapping("/emqx")
public class EmqxWebhook {
@PostMapping("/webhook")
public String webhook(@RequestBody HashMap req) {
log.info(req.toString());
if(req.get("action").equals("message_publish")){
String payload = req.get("payload").toString();
// TODO 发送事件至数据存储、WebSocket
}
return req.toString();
}
}
转发效果
分别启动 EMQX 服务、 SpringBoot 后端服务以及 Node.js 模拟的 MQTT 设备客户端，观察 EMQX 日志与后端服务日志。
小总结前面搭环境时也提到了，本次实践项目中仅用到了 EMQX 的设备接入、 WebHook 功能，但是 EMQX 本身还有很多强大的功能。映云科技基于 MQTT 协议研发出 EMQX 开源版， EMQX 企业版与 EMQX 云服务版，以及在边缘计算、流数据库方面的发力，构建起了面向物联网的现代数据基础设施生态， EMQX 是否值得你进一步探索？
Reference- https://www.emqx.io/docs/zh/v4.4/
If you have any questions or any bugs are found, please feel free to contact me.
Your comments and suggestions are welcome!
