init
This commit is contained in:
zhaohu
@@ -0,0 +1,24 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<parent>
|
||||
<groupId>org.springblade</groupId>
|
||||
<artifactId>blade-example</artifactId>
|
||||
<version>${revision}</version>
|
||||
</parent>
|
||||
|
||||
<artifactId>blade-mqtt-broker</artifactId>
|
||||
<name>${project.artifactId}</name>
|
||||
<packaging>jar</packaging>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>net.dreamlu</groupId>
|
||||
<artifactId>mica-mqttx-server-spring-boot-starter</artifactId>
|
||||
<version>3.1.12</version>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
</project>
|
||||
+47
@@ -0,0 +1,47 @@
|
||||
/**
|
||||
* BladeX Commercial License Agreement
|
||||
* Copyright (c) 2018-2099, https://bladex.cn. All rights reserved.
|
||||
* <p>
|
||||
* Use of this software is governed by the Commercial License Agreement
|
||||
* obtained after purchasing a license from BladeX.
|
||||
* <p>
|
||||
* 1. This software is for development use only under a valid license
|
||||
* from BladeX.
|
||||
* <p>
|
||||
* 2. Redistribution of this software's source code to any third party
|
||||
* without a commercial license is strictly prohibited.
|
||||
* <p>
|
||||
* 3. Licensees may copyright their own code but cannot use segments
|
||||
* from this software for such purposes. Copyright of this software
|
||||
* remains with BladeX.
|
||||
* <p>
|
||||
* Using this software signifies agreement to this License, and the software
|
||||
* must not be used for illegal purposes.
|
||||
* <p>
|
||||
* THIS SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY. The author is
|
||||
* not liable for any claims arising from secondary or illegal development.
|
||||
* <p>
|
||||
* Author: Chill Zhuang (bladejava@qq.com)
|
||||
*/
|
||||
package org.springblade.mqtt.broker;
|
||||
|
||||
import org.springblade.core.launch.BladeApplication;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
import org.springframework.scheduling.annotation.EnableScheduling;
|
||||
|
||||
import static org.springblade.common.constant.LauncherConstant.APPLICATION_MQTT_BROKER_NAME;
|
||||
|
||||
/**
|
||||
* MQTT Broker 启动器
|
||||
*
|
||||
* @author Chill
|
||||
*/
|
||||
@EnableScheduling
|
||||
@SpringBootApplication
|
||||
public class MqttBrokerApplication {
|
||||
|
||||
public static void main(String[] args) {
|
||||
BladeApplication.run(APPLICATION_MQTT_BROKER_NAME, MqttBrokerApplication.class, args);
|
||||
}
|
||||
|
||||
}
|
||||
+107
@@ -0,0 +1,107 @@
|
||||
/**
|
||||
* BladeX Commercial License Agreement
|
||||
* Copyright (c) 2018-2099, https://bladex.cn. All rights reserved.
|
||||
* <p>
|
||||
* Use of this software is governed by the Commercial License Agreement
|
||||
* obtained after purchasing a license from BladeX.
|
||||
* <p>
|
||||
* 1. This software is for development use only under a valid license
|
||||
* from BladeX.
|
||||
* <p>
|
||||
* 2. Redistribution of this software's source code to any third party
|
||||
* without a commercial license is strictly prohibited.
|
||||
* <p>
|
||||
* 3. Licensees may copyright their own code but cannot use segments
|
||||
* from this software for such purposes. Copyright of this software
|
||||
* remains with BladeX.
|
||||
* <p>
|
||||
* Using this software signifies agreement to this License, and the software
|
||||
* must not be used for illegal purposes.
|
||||
* <p>
|
||||
* THIS SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY. The author is
|
||||
* not liable for any claims arising from secondary or illegal development.
|
||||
* <p>
|
||||
* Author: Chill Zhuang (bladejava@qq.com)
|
||||
*/
|
||||
package org.springblade.mqtt.broker.listener;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import net.dreamlu.iot.mqtt.spring.server.event.MqttClientOfflineEvent;
|
||||
import net.dreamlu.iot.mqtt.spring.server.event.MqttClientOnlineEvent;
|
||||
import org.springframework.context.event.EventListener;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
/**
|
||||
* 客户端上下线监听器
|
||||
*
|
||||
* @author Chill
|
||||
*/
|
||||
@Slf4j
|
||||
@Service
|
||||
public class MqttConnectStatusListener {
|
||||
|
||||
@EventListener
|
||||
public void online(MqttClientOnlineEvent event) {
|
||||
log.info("MQTT客户端上线: {}", event);
|
||||
|
||||
// 客户端上线处理逻辑
|
||||
String clientId = event.getClientId();
|
||||
|
||||
log.info("客户端上线 - clientId: {}", clientId);
|
||||
|
||||
// 可以在这里添加上线处理逻辑，例如：
|
||||
// 1. 记录上线日志
|
||||
// 2. 更新设备状态
|
||||
// 3. 发送上线通知
|
||||
// 4. 统计在线数量等
|
||||
|
||||
handleClientOnline(clientId);
|
||||
}
|
||||
|
||||
@EventListener
|
||||
public void offline(MqttClientOfflineEvent event) {
|
||||
log.info("MQTT客户端下线: {}", event);
|
||||
|
||||
// 客户端下线处理逻辑
|
||||
String clientId = event.getClientId();
|
||||
|
||||
log.info("客户端下线 - clientId: {}", clientId);
|
||||
|
||||
// 可以在这里添加下线处理逻辑，例如：
|
||||
// 1. 记录下线日志
|
||||
// 2. 更新设备状态
|
||||
// 3. 发送下线通知
|
||||
// 4. 清理资源等
|
||||
|
||||
handleClientOffline(clientId);
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理客户端上线
|
||||
*
|
||||
* @param clientId 客户端ID
|
||||
*/
|
||||
private void handleClientOnline(String clientId) {
|
||||
try {
|
||||
// 添加具体的上线处理逻辑
|
||||
log.debug("执行客户端上线处理: clientId={}", clientId);
|
||||
} catch (Exception e) {
|
||||
log.error("处理客户端上线异常: clientId={}", clientId, e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理客户端下线
|
||||
*
|
||||
* @param clientId 客户端ID
|
||||
*/
|
||||
private void handleClientOffline(String clientId) {
|
||||
try {
|
||||
// 添加具体的下线处理逻辑
|
||||
log.debug("执行客户端下线处理: clientId={}", clientId);
|
||||
} catch (Exception e) {
|
||||
log.error("处理客户端下线异常: clientId={}", clientId, e);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
+91
@@ -0,0 +1,91 @@
|
||||
/**
|
||||
* BladeX Commercial License Agreement
|
||||
* Copyright (c) 2018-2099, https://bladex.cn. All rights reserved.
|
||||
* <p>
|
||||
* Use of this software is governed by the Commercial License Agreement
|
||||
* obtained after purchasing a license from BladeX.
|
||||
* <p>
|
||||
* 1. This software is for development use only under a valid license
|
||||
* from BladeX.
|
||||
* <p>
|
||||
* 2. Redistribution of this software's source code to any third party
|
||||
* without a commercial license is strictly prohibited.
|
||||
* <p>
|
||||
* 3. Licensees may copyright their own code but cannot use segments
|
||||
* from this software for such purposes. Copyright of this software
|
||||
* remains with BladeX.
|
||||
* <p>
|
||||
* Using this software signifies agreement to this License, and the software
|
||||
* must not be used for illegal purposes.
|
||||
* <p>
|
||||
* THIS SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY. The author is
|
||||
* not liable for any claims arising from secondary or illegal development.
|
||||
* <p>
|
||||
* Author: Chill Zhuang (bladejava@qq.com)
|
||||
*/
|
||||
package org.springblade.mqtt.broker.listener;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import net.dreamlu.iot.mqtt.codec.MqttPublishMessage;
|
||||
import net.dreamlu.iot.mqtt.codec.MqttQoS;
|
||||
import net.dreamlu.iot.mqtt.core.server.event.IMqttMessageListener;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.tio.core.ChannelContext;
|
||||
import org.tio.utils.buffer.ByteBufferUtil;
|
||||
|
||||
/**
|
||||
* 服务端接收消息监听器
|
||||
*
|
||||
* @author Chill
|
||||
*/
|
||||
@Slf4j
|
||||
@Service
|
||||
public class MqttServerMessageListener implements IMqttMessageListener {
|
||||
|
||||
@Override
|
||||
public void onMessage(ChannelContext context, String clientId, String topic, MqttQoS qoS, MqttPublishMessage message) {
|
||||
log.info("收到MQTT消息 - clientId: {}, topic: {}, qoS: {}, payload: {}",
|
||||
clientId, topic, qoS, ByteBufferUtil.toString(message.getPayload()));
|
||||
|
||||
// 自定义数据流转逻辑
|
||||
// 可以在这里添加消息处理逻辑，例如：
|
||||
// 1. 数据解析和格式化
|
||||
// 2. 业务规则处理
|
||||
// 3. 数据存储
|
||||
// 4. 消息转发
|
||||
// 5. 告警处理等
|
||||
|
||||
handleMessage(clientId, topic, message);
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理接收到的消息
|
||||
*
|
||||
* @param clientId 客户端ID
|
||||
* @param topic 主题
|
||||
* @param message 消息
|
||||
*/
|
||||
private void handleMessage(String clientId, String topic, MqttPublishMessage message) {
|
||||
try {
|
||||
String payload = ByteBufferUtil.toString(message.getPayload());
|
||||
|
||||
// 根据主题进行不同的处理
|
||||
if (topic.contains("/property/")) {
|
||||
log.info("处理属性消息: {}", payload);
|
||||
// 处理设备属性上报
|
||||
} else if (topic.contains("/event/")) {
|
||||
log.info("处理事件消息: {}", payload);
|
||||
// 处理设备事件上报
|
||||
} else if (topic.contains("/command/")) {
|
||||
log.info("处理命令消息: {}", payload);
|
||||
// 处理设备命令响应
|
||||
} else {
|
||||
log.info("处理通用消息: {}", payload);
|
||||
// 处理其他类型消息
|
||||
}
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("处理MQTT消息异常: clientId={}, topic={}", clientId, topic, e);
|
||||
}
|
||||
}
|
||||
}
|
||||
+49
@@ -0,0 +1,49 @@
|
||||
/**
|
||||
* BladeX Commercial License Agreement
|
||||
* Copyright (c) 2018-2099, https://bladex.cn. All rights reserved.
|
||||
* <p>
|
||||
* Use of this software is governed by the Commercial License Agreement
|
||||
* obtained after purchasing a license from BladeX.
|
||||
* <p>
|
||||
* 1. This software is for development use only under a valid license
|
||||
* from BladeX.
|
||||
* <p>
|
||||
* 2. Redistribution of this software's source code to any third party
|
||||
* without a commercial license is strictly prohibited.
|
||||
* <p>
|
||||
* 3. Licensees may copyright their own code but cannot use segments
|
||||
* from this software for such purposes. Copyright of this software
|
||||
* remains with BladeX.
|
||||
* <p>
|
||||
* Using this software signifies agreement to this License, and the software
|
||||
* must not be used for illegal purposes.
|
||||
* <p>
|
||||
* THIS SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY. The author is
|
||||
* not liable for any claims arising from secondary or illegal development.
|
||||
* <p>
|
||||
* Author: Chill Zhuang (bladejava@qq.com)
|
||||
*/
|
||||
package org.springblade.mqtt.broker.service;
|
||||
|
||||
import org.tio.core.ChannelContext;
|
||||
|
||||
/**
|
||||
* MQTT 认证服务
|
||||
*
|
||||
* @author Chill
|
||||
*/
|
||||
public interface IMqttAuthService {
|
||||
|
||||
/**
|
||||
* 设备连接认证
|
||||
*
|
||||
* @param context ChannelContext
|
||||
* @param clientId clientId
|
||||
* @param userName userName
|
||||
* @param password password
|
||||
* @param clientNodeIp clientNodeIp
|
||||
* @return 是否成功
|
||||
*/
|
||||
boolean hasAuth(ChannelContext context, String clientId, String userName, String password, String clientNodeIp);
|
||||
|
||||
}
|
||||
+67
@@ -0,0 +1,67 @@
|
||||
/**
|
||||
* BladeX Commercial License Agreement
|
||||
* Copyright (c) 2018-2099, https://bladex.cn. All rights reserved.
|
||||
* <p>
|
||||
* Use of this software is governed by the Commercial License Agreement
|
||||
* obtained after purchasing a license from BladeX.
|
||||
* <p>
|
||||
* 1. This software is for development use only under a valid license
|
||||
* from BladeX.
|
||||
* <p>
|
||||
* 2. Redistribution of this software's source code to any third party
|
||||
* without a commercial license is strictly prohibited.
|
||||
* <p>
|
||||
* 3. Licensees may copyright their own code but cannot use segments
|
||||
* from this software for such purposes. Copyright of this software
|
||||
* remains with BladeX.
|
||||
* <p>
|
||||
* Using this software signifies agreement to this License, and the software
|
||||
* must not be used for illegal purposes.
|
||||
* <p>
|
||||
* THIS SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY. The author is
|
||||
* not liable for any claims arising from secondary or illegal development.
|
||||
* <p>
|
||||
* Author: Chill Zhuang (bladejava@qq.com)
|
||||
*/
|
||||
package org.springblade.mqtt.broker.service.impl;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springblade.mqtt.broker.service.IMqttAuthService;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.tio.core.ChannelContext;
|
||||
|
||||
/**
|
||||
* MQTT 认证处理服务
|
||||
*
|
||||
* @author Chill
|
||||
*/
|
||||
@Slf4j
|
||||
@Service
|
||||
public class MqttAuthServiceImpl implements IMqttAuthService {
|
||||
|
||||
/**
|
||||
* MQTT 认证最大间隔时间，默认 5 分钟
|
||||
*/
|
||||
@Value("${mqtt.auth.timeSecondDiff:300}")
|
||||
private int timeSecondDiff;
|
||||
|
||||
@Override
|
||||
public boolean hasAuth(ChannelContext context, String clientId, String userName, String password, String clientNodeIp) {
|
||||
log.info("MQTT 客户端认证: clientId={}, userName={}, clientNodeIp={}", clientId, userName, clientNodeIp);
|
||||
|
||||
// 自定义账号认证逻辑
|
||||
// 这里可以根据实际需求添加具体的认证逻辑
|
||||
// 例如：查询数据库验证用户名密码、检查设备白名单等
|
||||
|
||||
// 简单的demo认证逻辑：允许所有连接
|
||||
if (userName != null && password != null) {
|
||||
log.info("MQTT 客户端认证成功: clientId={}", clientId);
|
||||
return true;
|
||||
}
|
||||
|
||||
log.warn("MQTT 客户端认证失败: clientId={}", clientId);
|
||||
return false;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,45 @@
|
||||
# 服务器配置
|
||||
server:
|
||||
port: 8290
|
||||
undertow:
|
||||
# 以下的配置会影响buffer,这些buffer会用于服务器连接的IO操作,有点类似netty的池化内存管理
|
||||
buffer-size: 1024
|
||||
# 是否分配的直接内存
|
||||
direct-buffers: true
|
||||
# 线程配置
|
||||
threads:
|
||||
# 设置IO线程数, 它主要执行非阻塞的任务,它们会负责多个连接, 默认设置每个CPU核心一个线程
|
||||
io: 16
|
||||
# 阻塞任务线程池, 当执行类似servlet请求阻塞操作, undertow会从这个线程池中取得线程,它的值设置取决于系统的负载
|
||||
worker: 400
|
||||
servlet:
|
||||
# 编码配置
|
||||
encoding:
|
||||
charset: UTF-8
|
||||
force: true
|
||||
|
||||
# MQTT服务端配置
|
||||
mqtt:
|
||||
server:
|
||||
enabled: true # 是否开启服务端，默认：true
|
||||
port: 11883 # 端口，默认：1883
|
||||
name: Blade-MQTT-Broker # 名称，默认：Mica-Mqtt-Server
|
||||
heartbeat-timeout: 120000 # 心跳超时，单位毫秒，默认: 1000 * 120
|
||||
read-buffer-size: 8KB # 接收数据的 buffer size，默认：8k
|
||||
max-bytes-in-message: 10MB # 消息解析最大 bytes 长度，默认：10M
|
||||
auth:
|
||||
enable: false # 是否开启 mqtt 认证
|
||||
debug: true # 如果开启 prometheus 指标收集建议关闭
|
||||
stat-enable: true # 开启指标收集，debug 和 prometheus 开启时需要打开，默认开启，关闭节省内存
|
||||
web-port: 28083 # http、websocket 端口，默认：8083
|
||||
websocket-enable: true # 是否开启 websocket，默认： true
|
||||
http-enable: false # 是否开启 http api，默认： false
|
||||
ssl: # mqtt tcp ssl 认证
|
||||
enabled: false # 是否开启 ssl 认证，2.1.0 开始支持双向认证
|
||||
keystore-path: # 必须参数：ssl keystore 目录，支持 classpath:/ 路径。
|
||||
keystore-pass: # 必选参数：ssl keystore 密码
|
||||
truststore-path: # 可选参数：ssl 双向认证 truststore 目录，支持 classpath:/ 路径。
|
||||
truststore-pass: # 可选参数：ssl 双向认证 truststore 密码
|
||||
client-auth: none # 是否需要客户端认证（双向认证），默认：NONE（不需要）
|
||||
auth:
|
||||
timeSecondDiff: 300 # 认证时间差，默认5分钟
|
||||
Reference in New Issue
Block a user