使用 EMQX 开源版的 Webhook 机制处理消息并存储数据

1、前言

EMQX 是一款强大的开源 MQTT 消息代理，它支持大量的连接和高吞吐量，适用于各种物联网应用。Webhook 是 EMQX 提供的扩展功能之一，用于将消息推送到外部的 HTTP 服务。在本文中，我们将介绍如何使用 EMQX 开源版的 Webhook 机制，并展示如何处理收到的 Webhook 请求，将其中的数据存储到数据库中。

2、Webhook 简介

Webhook 是一种常见的 HTTP 回调机制，用于将事件或数据推送到外部服务器。当 MQTT 客户端发布消息时，EMQX 可以通过 Webhook 将该消息发送给指定的 HTTP 端点，方便我们在接收到消息后进一步处理数据。

3、搭建 Webhook 服务

接下来，我们编写一个简单的 SpringBoot 2.7服务，用于接收 EMQX 的 Webhook 请求并将其中的数据存储到数据库中。

3.1、项目依赖

在 pom.xml 中添加以下依赖：

    <dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-core</artifactId><version>2.13.5</version></dependency><!-- Jackson Databind --><dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-databind</artifactId><version>2.13.5}</version></dependency><!-- Jackson Annotations --><dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-annotations</artifactId><version>2.13.5</version></dependency><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>8.0.33</version></dependency><dependency><groupId>com.baomidou</groupId><artifactId>mybatis-plus-boot-starter</artifactId><version>3.5.6</version></dependency></dependencies>

3.2、实现 Webhook 控制器

3.2.1、Controller

package ....这里填写你自己的import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.*;import java.io.IOException;
import java.util.Map;@RestController
@RequestMapping("/emqx/test")
@AllArgsConstructor
@Slf4j
public class WebhookController {private final EmqxTestService emqxTestService;private final ObjectMapper objectMapper = new ObjectMapper();@PostMapping("/webhook")public String webhook(@RequestBody String payload) {try {// 解析主 JSON 字符串为 MapMap<String, Object> payloadMap = objectMapper.readValue(payload, new TypeReference<Map<String, Object>>() {});// 从主 Map 中提取 clientid 和 topicString clientId = (String) payloadMap.get("clientid");String topic = (String) payloadMap.get("topic");log.info("Received clientid: {}", clientId);log.info("Received topic: {}", topic);// 提取 payload 字段的 JSON 字符串String payloadString = (String) payloadMap.get("payload");// 解析 payload 字段的 JSON 字符串为 MapMap<String, Object> payloadDataMap = objectMapper.readValue(payloadString, new TypeReference<Map<String, Object>>() {});// 从 payload 数据中提取 msg 参数String msg = (String) payloadDataMap.get("msg");log.info("Received msg: {}", msg);// 创建 EmqxTest 实例并设置字段EmqxTest testData = new EmqxTest();testData.setData(payload);testData.setClientId(clientId);testData.setTopic(topic);// 保存数据emqxTestService.insertData(testData);} catch (IOException e) {log.error("解析JSON有效负载失败", e);return "Error parsing payload";}return "Received";}
}

3.2.2、Service

package ....这里填写你自己的import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.ldb.tool.entity.EmqxTest;
import com.ldb.tool.mapper.EmqxTestMapper;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;import java.util.Date;
import java.util.List;@Service
@AllArgsConstructor
@Slf4j
public class EmqxTestService {private final EmqxTestMapper emqxTestMapper;public EmqxTest insertData(EmqxTest testData) {EmqxTest emqxTest = new EmqxTest();// 你可以手动设置其他需要的字段，如 clientId, topic, data 等emqxTest.setClientId(testData.getClientId());emqxTest.setTopic(testData.getTopic());emqxTest.setData(testData.getData());emqxTest.setCreateTime(new Date()); // 如果你有自动填充策略，可以忽略这行this.emqxTestMapper.insert(emqxTest);return emqxTest;}
}

3.2.3、Mapper

package ...这里填写你自己的;import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.ldb.tool.entity.EmqxTest;public interface EmqxTestMapper extends BaseMapper<EmqxTest> {
}

3.2.4、Entity

package ...这里填写你自己的;import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import lombok.Data;import java.io.Serializable;
import java.util.Date;@TableName("emqx_test")
@Data
public class EmqxTest implements Serializable {private static final long serialVersionUID = 1L;@TableId(value = "id", type = IdType.ASSIGN_ID)private Long id;private String clientId;private String topic;private String data;private Date createTime;private Date updateTime;
}

4、配置 EMQX Webhook

4.1、运行

我们这里使用docker来运行EMQX。

通过 Docker 运行 EMQX | EMQX文档

4.1.1、获取镜像

docker pull emqx/emqx:5.8.0

4.1.2、启动容器


docker run -d --name emqx \-p 1883:1883 -p 8083:8083 \-p 8084:8084 -p 8883:8883 \-p 18083:18083 \-v $PWD/data:/opt/emqx/data \-v $PWD/log:/opt/emqx/log \emqx/emqx:5.8.0