init
This commit is contained in:
yuhaiming
37
ruoyi-common/ruoyi-common-mqtt/pom.xml
Normal file
37
ruoyi-common/ruoyi-common-mqtt/pom.xml
Normal file
@@ -0,0 +1,37 @@
|
||||
<?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">
|
||||
<parent>
|
||||
<groupId>org.dromara</groupId>
|
||||
<artifactId>ruoyi-common</artifactId>
|
||||
<version>${revision}</version>
|
||||
</parent>
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<artifactId>ruoyi-common-mqtt</artifactId>
|
||||
|
||||
<description>
|
||||
ruoyi-common-mqtt mqtt模块
|
||||
</description>
|
||||
|
||||
<dependencies>
|
||||
<!-- 核心模块 -->
|
||||
<dependency>
|
||||
<groupId>org.dromara</groupId>
|
||||
<artifactId>ruoyi-common-core</artifactId>
|
||||
</dependency>
|
||||
|
||||
<!-- 序列化模块 -->
|
||||
<dependency>
|
||||
<groupId>org.dromara</groupId>
|
||||
<artifactId>ruoyi-common-json</artifactId>
|
||||
</dependency>
|
||||
|
||||
<!-- mqtt客户端 -->
|
||||
<dependency>
|
||||
<groupId>org.dromara.mica-mqtt</groupId>
|
||||
<artifactId>mica-mqtt-client-spring-boot-starter</artifactId>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
</project>
|
||||
@@ -0,0 +1,84 @@
|
||||
package org.dromara.common.mqtt.config;
|
||||
|
||||
import net.dreamlu.mica.net.utils.thread.ThreadUtils;
|
||||
import net.dreamlu.mica.net.utils.thread.pool.SynThreadPoolExecutor;
|
||||
import net.dreamlu.mica.net.utils.thread.pool.TioCallerRunsPolicy;
|
||||
import org.dromara.common.mqtt.listener.MqttClientConnectListener;
|
||||
import org.dromara.common.mqtt.listener.MqttClientGlobalMessageListener;
|
||||
import org.dromara.mica.mqtt.core.client.MqttClientCreator;
|
||||
import org.dromara.mica.mqtt.core.client.MqttClientCustomizer;
|
||||
import org.springframework.boot.autoconfigure.AutoConfiguration;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.core.task.VirtualThreadTaskExecutor;
|
||||
|
||||
import java.util.concurrent.LinkedBlockingQueue;
|
||||
import java.util.concurrent.ThreadFactory;
|
||||
import java.util.concurrent.ThreadPoolExecutor;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
/**
|
||||
* mqtt客户端配置初始化
|
||||
* <p>
|
||||
* 用法文档 <a href="https://mica-mqtt.dreamlu.net/guide/spring/client.html">...</a>
|
||||
* 测试server搭建:
|
||||
* 可执行下载其他mqtt服务端搭建
|
||||
* 也可使用 mica自带的server搭建 <a href="https://mica-mqtt.dreamlu.net/guide/spring/server.html">...</a>
|
||||
*
|
||||
* @author Lion Li
|
||||
*/
|
||||
@AutoConfiguration
|
||||
@ConditionalOnProperty(value = "mqtt.client.enabled", havingValue = "true")
|
||||
public class MqttAutoConfiguration {
|
||||
|
||||
/**
|
||||
* 创建 MQTT 客户端连接监听器。
|
||||
*
|
||||
* @param mqttClientCreator MQTT 客户端创建器
|
||||
* @return MQTT 客户端连接监听器
|
||||
*/
|
||||
@Bean
|
||||
public MqttClientConnectListener mqttClientConnectListener(MqttClientCreator mqttClientCreator) {
|
||||
return new MqttClientConnectListener(mqttClientCreator);
|
||||
}
|
||||
|
||||
/**
|
||||
* 创建 MQTT 全局消息监听器。
|
||||
*
|
||||
* @return MQTT 全局消息监听器
|
||||
*/
|
||||
@Bean
|
||||
public MqttClientGlobalMessageListener mqttClientGlobalMessageListener() {
|
||||
return new MqttClientGlobalMessageListener();
|
||||
}
|
||||
|
||||
/**
|
||||
* 客户端使用虚拟线程配置
|
||||
*/
|
||||
@Bean
|
||||
public MqttClientCustomizer mqttClientCustomizer() {
|
||||
return creator -> {
|
||||
// 这个数不重要 已经使用虚拟线程 就是填一下防止报错
|
||||
int corePoolSize = ThreadUtils.CORE_POOL_SIZE;
|
||||
|
||||
ThreadFactory factory = new VirtualThreadTaskExecutor("tio-worker-virtual").getVirtualThreadFactory();
|
||||
SynThreadPoolExecutor tioExecutor = new SynThreadPoolExecutor(corePoolSize, corePoolSize,
|
||||
0L, new LinkedBlockingQueue<>(), factory, new TioCallerRunsPolicy());
|
||||
tioExecutor.prestartCoreThread();
|
||||
creator.tioExecutor(tioExecutor);
|
||||
|
||||
ThreadFactory factory1 = new VirtualThreadTaskExecutor("tio-group-virtual").getVirtualThreadFactory();
|
||||
ThreadPoolExecutor groupExecutor = new ThreadPoolExecutor(corePoolSize, corePoolSize,
|
||||
0L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(), factory1, new TioCallerRunsPolicy());
|
||||
groupExecutor.prestartCoreThread();
|
||||
creator.groupExecutor(groupExecutor);
|
||||
|
||||
ThreadFactory factory2 = new VirtualThreadTaskExecutor("biz-worker-virtual").getVirtualThreadFactory();
|
||||
ThreadPoolExecutor mqttExecutor = new ThreadPoolExecutor(corePoolSize, corePoolSize,
|
||||
0L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(), factory2, new TioCallerRunsPolicy());
|
||||
mqttExecutor.prestartCoreThread();
|
||||
creator.mqttExecutor(mqttExecutor);
|
||||
};
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,56 @@
|
||||
package org.dromara.common.mqtt.listener;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import net.dreamlu.mica.net.core.ChannelContext;
|
||||
import org.dromara.mica.mqtt.core.client.IMqttClientConnectListener;
|
||||
import org.dromara.mica.mqtt.core.client.MqttClientCreator;
|
||||
|
||||
/**
|
||||
* 客户端连接状态监听
|
||||
*
|
||||
* @author Lion Li
|
||||
*/
|
||||
@Slf4j
|
||||
public class MqttClientConnectListener implements IMqttClientConnectListener {
|
||||
|
||||
private final MqttClientCreator mqttClientCreator;
|
||||
|
||||
/**
|
||||
* 创建 MQTT 客户端连接监听器。
|
||||
*
|
||||
* @param mqttClientCreator MQTT 客户端创建器
|
||||
*/
|
||||
public MqttClientConnectListener(MqttClientCreator mqttClientCreator) {
|
||||
this.mqttClientCreator = mqttClientCreator;
|
||||
}
|
||||
|
||||
/**
|
||||
* 客户端连接成功回调。
|
||||
*
|
||||
* @param context 通道上下文
|
||||
* @param isReconnect 是否重连
|
||||
*/
|
||||
@Override
|
||||
public void onConnected(ChannelContext context, boolean isReconnect) {
|
||||
// 创建连接
|
||||
log.info("MqttConnectedEvent:{}", context);
|
||||
}
|
||||
|
||||
/**
|
||||
* 客户端断开连接回调。
|
||||
*
|
||||
* @param context 通道上下文
|
||||
* @param throwable 断开异常
|
||||
* @param remark 断开说明
|
||||
* @param isRemove 是否移除连接
|
||||
*/
|
||||
@Override
|
||||
public void onDisconnect(ChannelContext context, Throwable throwable, String remark, boolean isRemove) {
|
||||
// 离线时更新重连
|
||||
log.info("MqttDisconnectEvent:{}", context, throwable);
|
||||
// 在断线时更新 clientId、username、password
|
||||
// mqttClientCreator.clientId("newClient" + System.currentTimeMillis())
|
||||
// .username("newUserName")
|
||||
// .password("newPassword");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,31 @@
|
||||
package org.dromara.common.mqtt.listener;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import net.dreamlu.mica.net.core.ChannelContext;
|
||||
import org.dromara.mica.mqtt.codec.message.MqttPublishMessage;
|
||||
import org.dromara.mica.mqtt.core.client.IMqttClientGlobalMessageListener;
|
||||
|
||||
import java.nio.charset.StandardCharsets;
|
||||
|
||||
/**
|
||||
* 全局消息监听，可以监听到所有订阅消息
|
||||
*
|
||||
* @author Lion Li
|
||||
*/
|
||||
@Slf4j
|
||||
public class MqttClientGlobalMessageListener implements IMqttClientGlobalMessageListener {
|
||||
|
||||
/**
|
||||
* 接收并记录全局 MQTT 消息。
|
||||
*
|
||||
* @param context 通道上下文
|
||||
* @param topic 主题
|
||||
* @param message MQTT 消息对象
|
||||
* @param payload 消息载荷
|
||||
*/
|
||||
@Override
|
||||
public void onMessage(ChannelContext context, String topic, MqttPublishMessage message, byte[] payload) {
|
||||
log.info("MqttGlobalMessageEvent => topic: {}, msg: {}", topic, new String(payload, StandardCharsets.UTF_8));
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
org.dromara.common.mqtt.config.MqttAutoConfiguration
|
||||
Reference in New Issue
Block a user