Add - Mqtt broker
parent
5b80094a3f
commit
372bbee821
@ -0,0 +1,136 @@
|
||||
<?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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<!--<parent>
|
||||
<groupId>com.ruoyi</groupId>
|
||||
<artifactId>ruoyi-modules</artifactId>
|
||||
<version>3.6.3</version>
|
||||
</parent>-->
|
||||
<parent>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-parent</artifactId>
|
||||
<version>2.7.9</version>
|
||||
<relativePath/> <!-- lookup parent from repository -->
|
||||
</parent>
|
||||
|
||||
<artifactId>hw-mqtt-broker</artifactId>
|
||||
|
||||
<description>MQTT Broker</description>
|
||||
|
||||
<properties>
|
||||
<revision>2.1.0</revision>
|
||||
<java.version>1.8</java.version>
|
||||
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
|
||||
<mica.version>2.7.9</mica.version>
|
||||
<spring.boot.version>2.7.9</spring.boot.version>
|
||||
<fastjson.version>1.2.83</fastjson.version>
|
||||
<tinylog.version>2.6.0</tinylog.version>
|
||||
<junit-jupiter.version>5.9.2</junit-jupiter.version>
|
||||
</properties>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>net.dreamlu</groupId>
|
||||
<artifactId>mica-mqtt-server-spring-boot-starter</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-web</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>net.dreamlu</groupId>
|
||||
<artifactId>mica-lite</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>net.dreamlu</groupId>
|
||||
<artifactId>mica-logging</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>net.dreamlu</groupId>
|
||||
<artifactId>mica-redis</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>net.dreamlu</groupId>
|
||||
<artifactId>mica-openapi</artifactId>
|
||||
</dependency>
|
||||
<!-- 开启 prometheus 指标收集，详见： http://localhost:30012/actuator/prometheus -->
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-actuator</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>io.micrometer</groupId>
|
||||
<artifactId>micrometer-registry-prometheus</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.projectlombok</groupId>
|
||||
<artifactId>lombok</artifactId>
|
||||
<scope>provided</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
<dependencyManagement>
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-dependencies</artifactId>
|
||||
<version>${spring.boot.version}</version>
|
||||
<type>pom</type>
|
||||
<scope>import</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>net.dreamlu</groupId>
|
||||
<artifactId>mica-bom</artifactId>
|
||||
<version>${mica.version}</version>
|
||||
<type>pom</type>
|
||||
<scope>import</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>com.alibaba</groupId>
|
||||
<artifactId>fastjson</artifactId>
|
||||
<version>${fastjson.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>net.dreamlu</groupId>
|
||||
<artifactId>mica-mqtt-server-spring-boot-starter</artifactId>
|
||||
<version>${revision}</version>
|
||||
</dependency>
|
||||
<!-- tinylog 内存占用更小、性能更好，适合边缘设备 -->
|
||||
<dependency>
|
||||
<groupId>org.tinylog</groupId>
|
||||
<artifactId>slf4j-tinylog</artifactId>
|
||||
<version>${tinylog.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.tinylog</groupId>
|
||||
<artifactId>tinylog-impl</artifactId>
|
||||
<version>${tinylog.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.junit.jupiter</groupId>
|
||||
<artifactId>junit-jupiter-engine</artifactId>
|
||||
<version>${junit-jupiter.version}</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
</dependencyManagement>
|
||||
|
||||
<build>
|
||||
<finalName>${project.artifactId}</finalName>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-maven-plugin</artifactId>
|
||||
<executions>
|
||||
<execution>
|
||||
<goals>
|
||||
<goal>repackage</goal>
|
||||
</goals>
|
||||
</execution>
|
||||
</executions>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
|
||||
</project>
|
||||
@ -0,0 +1,13 @@
|
||||
package com.hw.mqtt;
|
||||
|
||||
import org.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
|
||||
@SpringBootApplication
|
||||
public class HwMqttBrokerApplication {
|
||||
|
||||
public static void main(String[] args) {
|
||||
SpringApplication.run(HwMqttBrokerApplication.class, args);
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,22 @@
|
||||
package com.hw.mqtt.auth;
|
||||
|
||||
import net.dreamlu.iot.mqtt.core.server.auth.IMqttServerAuthHandler;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.tio.core.ChannelContext;
|
||||
|
||||
/**
|
||||
* mqtt tcp、websocket 认证
|
||||
* @author WenJY
|
||||
* @date 2023-03-14 12:19
|
||||
* @return null
|
||||
*/
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
public class MqttAuthHandler implements IMqttServerAuthHandler {
|
||||
|
||||
@Override
|
||||
public boolean authenticate(ChannelContext context, String uniqueId, String clientId, String userName, String password) {
|
||||
// 客户端认证逻辑实现
|
||||
return true;
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,37 @@
|
||||
package com.hw.mqtt.auth;
|
||||
|
||||
import net.dreamlu.iot.mqtt.core.server.http.api.code.ResultCode;
|
||||
import net.dreamlu.iot.mqtt.core.server.http.api.result.Result;
|
||||
import net.dreamlu.iot.mqtt.core.server.http.handler.HttpFilter;
|
||||
import net.dreamlu.iot.mqtt.core.server.http.handler.MqttHttpRoutes;
|
||||
import org.springframework.beans.factory.InitializingBean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.tio.http.common.HttpRequest;
|
||||
import org.tio.http.common.HttpResponse;
|
||||
|
||||
/**
|
||||
* mqtt http 接口认证
|
||||
* @author WenJY
|
||||
* @date 2023-03-14 12:20
|
||||
* @return null
|
||||
*/
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
public class MqttHttpAuthFilter implements HttpFilter, InitializingBean {
|
||||
|
||||
@Override
|
||||
public boolean filter(HttpRequest request) throws Exception {
|
||||
// 自行实现逻辑
|
||||
return true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public HttpResponse response(HttpRequest request) {
|
||||
// 认证不通过时的响应
|
||||
return Result.fail(request, ResultCode.E103);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void afterPropertiesSet() throws Exception {
|
||||
MqttHttpRoutes.addFilter(this);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,25 @@
|
||||
package com.hw.mqtt.domain;
|
||||
|
||||
import lombok.AllArgsConstructor;
|
||||
import lombok.Data;
|
||||
import lombok.NoArgsConstructor;
|
||||
|
||||
/**
|
||||
* @author WenJY
|
||||
* @date 2023年03月14日 13:46
|
||||
*/
|
||||
@Data
|
||||
@NoArgsConstructor
|
||||
@AllArgsConstructor
|
||||
public class ServerNode {
|
||||
|
||||
/**
|
||||
* 节点名称
|
||||
*/
|
||||
private String name;
|
||||
/**
|
||||
* ip:port
|
||||
*/
|
||||
private String peerHost;
|
||||
|
||||
}
|
||||
@ -0,0 +1,40 @@
|
||||
package com.hw.mqtt.listener;
|
||||
|
||||
import net.dreamlu.iot.mqtt.codec.ByteBufferUtil;
|
||||
import net.dreamlu.iot.mqtt.codec.MqttPublishMessage;
|
||||
import net.dreamlu.iot.mqtt.codec.MqttQoS;
|
||||
import net.dreamlu.iot.mqtt.core.server.event.IMqttMessageListener;
|
||||
import net.dreamlu.iot.mqtt.spring.server.MqttServerTemplate;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.beans.factory.SmartInitializingSingleton;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.context.ApplicationContext;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.tio.core.ChannelContext;
|
||||
|
||||
/**
|
||||
* 消息监听器
|
||||
* @author WenJY
|
||||
* @date 2023-03-14 12:18
|
||||
* @param null
|
||||
* @return null
|
||||
*/
|
||||
@Service
|
||||
public class MqttServerMessageListener implements IMqttMessageListener, SmartInitializingSingleton {
|
||||
private static final Logger logger = LoggerFactory.getLogger(MqttServerMessageListener.class);
|
||||
@Autowired
|
||||
private ApplicationContext applicationContext;
|
||||
private MqttServerTemplate mqttServerTemplate;
|
||||
|
||||
@Override
|
||||
public void onMessage(ChannelContext context, String clientId, String topic, MqttQoS qos, MqttPublishMessage message) {
|
||||
logger.info("context:{} clientId:{} message:{} payload:{}", context, clientId, message, ByteBufferUtil.toString(message.getPayload()));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void afterSingletonsInstantiated() {
|
||||
// 单利 bean 初始化完成之后从 ApplicationContext 中获取 bean
|
||||
mqttServerTemplate = applicationContext.getBean(MqttServerTemplate.class);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,62 @@
|
||||
/*
|
||||
* Copyright (c) 2019-2029, Dreamlu 卢春梦 (596392912@qq.com & dreamlu.net).
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package com.hw.mqtt.service;
|
||||
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* mqtt broker 服务
|
||||
*
|
||||
* @author L.cm
|
||||
*/
|
||||
public interface IMqttBrokerService {
|
||||
|
||||
/**
|
||||
* 获取在线客户端数量
|
||||
* @author WenJY
|
||||
* @date 2023-03-14 13:51
|
||||
* @return long
|
||||
*/
|
||||
long getOnlineClientSize();
|
||||
|
||||
/**
|
||||
* 获取所有在线的客户端
|
||||
* @author WenJY
|
||||
* @date 2023-03-14 13:51
|
||||
* @return java.util.List<java.lang.String>
|
||||
*/
|
||||
List<String> getOnlineClients();
|
||||
|
||||
/**
|
||||
* 向指定主题发送消息
|
||||
* @author WenJY
|
||||
* @date 2023-03-14 13:50
|
||||
* @param topic
|
||||
* @param payload
|
||||
* @return boolean
|
||||
*/
|
||||
boolean publish(String topic,String payload);
|
||||
|
||||
/**
|
||||
* 主动关闭指定客户端连接
|
||||
* @author WenJY
|
||||
* @date 2023-03-18 9:23
|
||||
* @param clientId
|
||||
*/
|
||||
boolean closeClientById(String clientId);
|
||||
}
|
||||
@ -0,0 +1,29 @@
|
||||
package com.hw.mqtt.service;
|
||||
|
||||
import net.dreamlu.iot.mqtt.spring.server.MqttServerTemplate;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import java.nio.charset.StandardCharsets;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author WenJY
|
||||
* @date 2023-03-14 12:19
|
||||
* @param null
|
||||
* @return null
|
||||
*/
|
||||
@Service
|
||||
public class ServerService {
|
||||
private static final Logger logger = LoggerFactory.getLogger(ServerService.class);
|
||||
@Autowired
|
||||
private MqttServerTemplate server;
|
||||
|
||||
public boolean publish(String body) {
|
||||
boolean result = server.publishAll("/test/message", body.getBytes(StandardCharsets.UTF_8));
|
||||
logger.info("Mqtt publishAll result:{};payload:{}", result,body);
|
||||
return result;
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,96 @@
|
||||
/*
|
||||
* Copyright (c) 2019-2029, Dreamlu 卢春梦 (596392912@qq.com & dreamlu.net).
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package com.hw.mqtt.service.impl;
|
||||
|
||||
import com.hw.mqtt.enums.RedisKeys;
|
||||
import com.hw.mqtt.service.IMqttBrokerService;
|
||||
import net.dreamlu.iot.mqtt.spring.server.MqttServerTemplate;
|
||||
import net.dreamlu.mica.core.utils.StringPool;
|
||||
import net.dreamlu.mica.redis.cache.MicaRedisCache;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.Set;
|
||||
|
||||
/**
|
||||
* mqtt broker 服务
|
||||
*
|
||||
* @author L.cm
|
||||
*/
|
||||
@Service
|
||||
public class MqttBrokerServiceImpl implements IMqttBrokerService {
|
||||
private static final Logger logger = LoggerFactory.getLogger(MqttBrokerServiceImpl.class);
|
||||
@Autowired private MicaRedisCache redisCache;
|
||||
@Autowired private MqttServerTemplate server;
|
||||
|
||||
@Override
|
||||
public long getOnlineClientSize() {
|
||||
Set<String> keySet = redisCache.scan(RedisKeys.CONNECT_STATUS.getKey(StringPool.STAR));
|
||||
if (keySet.isEmpty()) {
|
||||
return 0L;
|
||||
}
|
||||
long result = 0;
|
||||
for (String redisKey : keySet) {
|
||||
Long count = redisCache.getSetOps().size(redisKey);
|
||||
if (count != null) {
|
||||
result += count;
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<String> getOnlineClients() {
|
||||
Set<String> keySet = redisCache.scan(RedisKeys.CONNECT_STATUS.getKey(StringPool.STAR));
|
||||
if (keySet.isEmpty()) {
|
||||
return Collections.emptyList();
|
||||
}
|
||||
List<String> clientList = new ArrayList<>();
|
||||
for (String redisKey : keySet) {
|
||||
Set<String> members = redisCache.sMembers(redisKey);
|
||||
if (members != null && !members.isEmpty()) {
|
||||
clientList.addAll(members);
|
||||
}
|
||||
}
|
||||
return clientList;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean publish(String topic, String payload) {
|
||||
boolean result = server.publishAll(topic, payload.getBytes(StandardCharsets.UTF_8));
|
||||
logger.info("Mqtt publishAll result:{};topic:{};payload:{}", result, topic, payload);
|
||||
return result;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean closeClientById(String clientId) {
|
||||
try{
|
||||
server.close(clientId);
|
||||
}catch (Exception ex){
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
@ -0,0 +1,40 @@
|
||||
/*
|
||||
* Copyright (c) 2019-2029, Dreamlu 卢春梦 (596392912@qq.com & dreamlu.net).
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package com.hw.mqtt.util;
|
||||
|
||||
import net.dreamlu.mica.core.utils.CharPool;
|
||||
|
||||
/**
|
||||
* redis 工具
|
||||
*
|
||||
* @author L.cm
|
||||
*/
|
||||
public class RedisUtil {
|
||||
|
||||
/**
|
||||
* 转换成 redis 的 pattern 规则
|
||||
*
|
||||
* @return pattern
|
||||
*/
|
||||
public static String getTopicPattern(String topicFilter) {
|
||||
// mqtt 分享主题 $share/{ShareName}/{filter}
|
||||
return topicFilter
|
||||
.replace(CharPool.PLUS, CharPool.STAR)
|
||||
.replace(CharPool.HASH, CharPool.STAR);
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,33 @@
|
||||
|
||||
spring:
|
||||
application:
|
||||
name: Mqtt-Broker
|
||||
profiles:
|
||||
active: server
|
||||
springdoc:
|
||||
swagger-ui:
|
||||
urls:
|
||||
- name: swagger
|
||||
url: /v3/api-docs
|
||||
# actuator management
|
||||
management:
|
||||
info:
|
||||
defaults:
|
||||
enabled: true
|
||||
metrics:
|
||||
tags:
|
||||
application: ${spring.application.name}
|
||||
endpoint:
|
||||
health:
|
||||
show-details: ALWAYS
|
||||
prometheus:
|
||||
enabled: true
|
||||
endpoints:
|
||||
web:
|
||||
exposure:
|
||||
include: '*'
|
||||
logging:
|
||||
level:
|
||||
root: info
|
||||
server: info # t-io ???????
|
||||
org.tio: info # t-io ???????
|
||||
@ -0,0 +1,10 @@
|
||||
|
||||
${AnsiColor.RED} ## ## ####### ######## ########
|
||||
${AnsiColor.RED} ### ### ## ## ## ##
|
||||
${AnsiColor.RED} #### #### ## ## ## ##
|
||||
${AnsiColor.RED} ## ### ## ## ## ## ##
|
||||
${AnsiColor.RED} ## ## ## ## ## ## ##
|
||||
${AnsiColor.RED} ## ## ## ## ## ##
|
||||
${AnsiColor.RED} ## ## ##### ## ## ##
|
||||
|
||||
${AnsiColor.BRIGHT_BLUE}:: ${spring.application.name} :: Running Spring Boot ${spring-boot.version} 🏃🏃🏃 ${AnsiColor.DEFAULT}
|
||||
Loading…
Reference in New Issue