XA/mqtt-car
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

去除无用示例

Browse Source
This commit is contained in:
zc
2025-12-08 10:48:38 +08:00
parent 871ae8be0a
commit af93d773ca
58 changed files with 2 additions and 2472 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
example/mica-mqtt-client-solon-plugin-example/pom.xml
View File

@@ -1,33 +0,0 @@
<?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.dromara.mica-mqtt</groupId>
<artifactId>example</artifactId>
<version>${revision}</version>
</parent>
<artifactId>mica-mqtt-client-solon-plugin-example</artifactId>
<dependencies>
<dependency>
<groupId>org.noear</groupId>
<artifactId>solon-web</artifactId>
</dependency>
<dependency>
<groupId>org.dromara.mica-mqtt</groupId>
<artifactId>mica-mqtt-client-solon-plugin</artifactId>
</dependency>
<dependency>
<groupId>org.noear</groupId>
<artifactId>solon-logging-simple</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<scope>provided</scope>
</dependency>
</dependencies>
</project>

19
example/mica-mqtt-client-solon-plugin-example/src/main/java/org/dromara/mica/mqtt/client/solon/MqttClientApplication.java
View File

@@ -1,19 +0,0 @@
package org.dromara.mica.mqtt.client.solon;
import org.noear.solon.Solon;
import org.noear.solon.annotation.Configuration;
/**
* @author wsq
*/
@Configuration
public class MqttClientApplication {
/**
* 先启动 mica-mqtt-server-spring-boot-example 再启动本项目，进行测试
*/
public static void main(String[] args) {
Solon.start(MqttClientApplication.class, args);
}
}

26
example/mica-mqtt-client-solon-plugin-example/src/main/java/org/dromara/mica/mqtt/client/solon/controller/ClientController.java
View File

@@ -1,26 +0,0 @@
package org.dromara.mica.mqtt.client.solon.controller;
import org.dromara.mica.mqtt.client.solon.service.ClientService;
import org.noear.solon.annotation.*;
@Mapping("/mqtt/client")
@Controller
public class ClientController {
@Inject
private ClientService service;
@Post
@Mapping("/publish")
public boolean publish(String body) {
return service.publish(body);
}
@Get
@Mapping("/sub")
public boolean sub() {
return service.sub();
}
}

41
example/mica-mqtt-client-solon-plugin-example/src/main/java/org/dromara/mica/mqtt/client/solon/listener/MqttClientConnectedListener.java
View File

@@ -1,41 +0,0 @@
/*
* 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 org.dromara.mica.mqtt.client.solon.listener;
import lombok.extern.slf4j.Slf4j;
import org.dromara.mica.mqtt.client.solon.event.MqttConnectedEvent;
import org.dromara.mica.mqtt.core.client.MqttClientCreator;
import org.noear.solon.annotation.Component;
import org.noear.solon.annotation.Inject;
import org.noear.solon.core.event.EventListener;
/**
* 客户端连接状态监听
*
* @author L.cm
*/
@Slf4j
@Component
public class MqttClientConnectedListener implements EventListener<MqttConnectedEvent> {
@Inject
private MqttClientCreator mqttClientCreator;
@Override
public void onEvent(MqttConnectedEvent mqttConnectedEvent) throws Throwable {
log.info("MqttConnectedEvent:{}", mqttConnectedEvent);
}
}

45
example/mica-mqtt-client-solon-plugin-example/src/main/java/org/dromara/mica/mqtt/client/solon/listener/MqttClientDisconnectListener.java
View File

@@ -1,45 +0,0 @@
/*
* 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 org.dromara.mica.mqtt.client.solon.listener;
import lombok.extern.slf4j.Slf4j;
import org.dromara.mica.mqtt.client.solon.event.MqttDisconnectEvent;
import org.dromara.mica.mqtt.core.client.MqttClientCreator;
import org.noear.solon.annotation.Component;
import org.noear.solon.annotation.Inject;
import org.noear.solon.core.event.EventListener;
/**
* 客户端连接状态监听
*
* @author L.cm
*/
@Slf4j
@Component
public class MqttClientDisconnectListener implements EventListener<MqttDisconnectEvent> {
@Inject
private MqttClientCreator mqttClientCreator;
@Override
public void onEvent(MqttDisconnectEvent mqttDisconnectEvent) throws Throwable {
log.info("MqttDisconnectEvent:{}", mqttDisconnectEvent);
// 在断线时更新 clientId、username、password，只能改这 3 个，不可调用其他方法。
// mqttClientCreator.clientId("newClient" + System.currentTimeMillis())
// .username("newUserName")
// .password("newPassword");
}
}

25
example/mica-mqtt-client-solon-plugin-example/src/main/java/org/dromara/mica/mqtt/client/solon/listener/MqttClientMessageListener.java
View File

@@ -1,25 +0,0 @@
package org.dromara.mica.mqtt.client.solon.listener;
import lombok.extern.slf4j.Slf4j;
import org.dromara.mica.mqtt.core.annotation.MqttClientSubscribe;
import org.dromara.mica.mqtt.codec.message.MqttPublishMessage;
import org.dromara.mica.mqtt.core.client.IMqttClientMessageListener;
import org.tio.core.ChannelContext;
import java.nio.charset.StandardCharsets;
/**
* 客户端消息监听的另一种方式
*
* @author L.cm
*/
@Slf4j
@MqttClientSubscribe("${topic1}")
public class MqttClientMessageListener implements IMqttClientMessageListener {
@Override
public void onMessage(ChannelContext context, String topic, MqttPublishMessage message, byte[] payload) {
log.info("MqttClientMessageListener,topic:{} payload:{}", topic, new String(payload, StandardCharsets.UTF_8));
}
}

68
example/mica-mqtt-client-solon-plugin-example/src/main/java/org/dromara/mica/mqtt/client/solon/listener/MqttClientSubscribeListener.java
View File

@@ -1,68 +0,0 @@
package org.dromara.mica.mqtt.client.solon.listener;
import lombok.extern.slf4j.Slf4j;
import org.dromara.mica.mqtt.core.annotation.MqttClientSubscribe;
import org.dromara.mica.mqtt.client.solon.pojo.User;
import org.dromara.mica.mqtt.codec.MqttQoS;
import org.dromara.mica.mqtt.codec.message.MqttPublishMessage;
import org.dromara.mica.mqtt.core.deserialize.MqttJsonDeserializer;
import org.noear.solon.annotation.Component;
import java.nio.charset.StandardCharsets;
import java.util.Map;
/**
* 客户端消息监听
*
* @author L.cm
*/
@Slf4j
@Component
public class MqttClientSubscribeListener {
@MqttClientSubscribe("/test/#")
public void subQos0(String topic, byte[] payload) {
log.info("MqttClientSubscribeListener.subQos0,topic:{} payload:{}", topic, new String(payload, StandardCharsets.UTF_8));
}
@MqttClientSubscribe(value = "/qos1/#", qos = MqttQoS.QOS1)
public void subQos1(String topic, byte[] payload) {
log.info("topic:{} payload:{}", topic, new String(payload, StandardCharsets.UTF_8));
}
@MqttClientSubscribe(
value = "/test/json",
deserialize = MqttJsonDeserializer.class // 2.4.5 开始支持 自定义序列化，默认 json 序列化
)
public void testJson(String topic, MqttPublishMessage message, Map<String, Object> data) {
// solon 插件为 2.4.6 开始支持，支持 2 到 3 个参数，字段类型映射规则（顺序）如下
// String 字符串会默认映射到 topic，
// Map<String, String> topicVars 会默认映射到 topic 中的变量解析（v2.5.4支持），注意：别跟消息序列化的冲突，消息反序列化不要用 Map<String, String>
// MqttPublishMessage 会默认映射到 原始的消息，可以拿到 mqtt5 的 props 参数
// byte[] 会映射到 mqtt 消息内容 payload
// ByteBuffer 会映射到 mqtt 消息内容 payload
// 其他类型会走序列化，确保消息能够序列化，默认为 json 序列化
log.info("topic:{} json data:{}", topic, data);
}
@MqttClientSubscribe(value = "/test/object")
public void testJson(String topic, MqttPublishMessage message, User<User> data) {
log.info("topic:{} json data:{}", topic, data);
}
/**
* 订阅，参数为可选，但是参数数量必须大于 2，
*
* @param topic topic 参数，可选参数
* @param topicVars 订阅 topic 模板 ${productKey} 中的变量解析（v2.5.4支持），可选参数，注意：类型必须为 Map<String, String>
* @param payload 消息内容
*/
@MqttClientSubscribe("/sys/${productKey}/${deviceName}/thing/sub/register")
public void thingSubRegister(String topic, Map<String, String> topicVars, byte[] payload) {
// 1.3.8 开始支持，@MqttClientSubscribe 注解支持 ${} 变量替换，会默认替换成 +
// 注意：mica-mqtt 会先从 Spring boot 配置中替换参数 ${}，如果存在配置会优先被替换。
log.info("topic:{} payload:{}", topic, new String(payload, StandardCharsets.UTF_8));
}
}

9
example/mica-mqtt-client-solon-plugin-example/src/main/java/org/dromara/mica/mqtt/client/solon/pojo/User.java
View File

@@ -1,9 +0,0 @@
package org.dromara.mica.mqtt.client.solon.pojo;
import lombok.Data;
@Data
public class User<T> {
private String name;
private T girlfriend;
}

32
example/mica-mqtt-client-solon-plugin-example/src/main/java/org/dromara/mica/mqtt/client/solon/service/ClientService.java
View File

@@ -1,32 +0,0 @@
package org.dromara.mica.mqtt.client.solon.service;
import lombok.extern.slf4j.Slf4j;
import org.dromara.mica.mqtt.client.solon.MqttClientTemplate;
import org.noear.solon.annotation.Component;
import org.noear.solon.annotation.Inject;
import java.nio.charset.StandardCharsets;
/**
* @author wsq
*/
@Slf4j
@Component
public class ClientService {
@Inject
private MqttClientTemplate client;
public boolean publish(String body) {
client.publish("/test/client", body.getBytes(StandardCharsets.UTF_8));
return true;
}
public boolean sub() {
client.subQos0("/test/#", (context, topic, message, payload) -> {
log.info("{}\t{}", topic, new String(payload, StandardCharsets.UTF_8));
});
return true;
}
}

36
example/mica-mqtt-client-solon-plugin-example/src/main/resources/app.yml
View File

@@ -1,36 +0,0 @@
server:
port: 30303
# solon 配置
solon:
logging:
appender:
console:
level: INFO
# mqtt-client 配置
mqtt:
client:
enabled: true # 是否开启客户端，默认：true
ip: 127.0.0.1 # 连接的服务端 ip ，默认：127.0.0.1
port: 1883 # 端口：默认：1883
name: Mica-Mqtt-Client # 名称，默认：Mica-Mqtt-Client
clientId: 000001 # 客户端Id（非常重要，一般为设备 sn，不可重复）
username: mica # 认证的用户名，注意：2.5.x 开始将 user-name 改成了 username
password: mica # 认证的密码
timeout: 5 # 超时时间，单位：秒，默认：5秒
reconnect: true # 是否重连，默认：true
re-interval: 5000 # 重连时间，默认 5000 毫秒
version: mqtt_3_1_1 # mqtt 协议版本，可选 MQTT_3_1、mqtt_3_1_1、mqtt_5，默认：mqtt_3_1_1
read-buffer-size: 8KB # 接收数据的 buffer size，默认：8k
max-bytes-in-message: 10MB # 消息解析最大 bytes 长度，默认：10M
keep-alive-secs: 60 # keep-alive 时间，单位：秒
heartbeat-mode: LAST_REQ # 心跳模式，支持最后发送或接收心跳时间来计算心跳，默认：最后发送心跳的时间。（2.4.3 开始支持）
heartbeat-timeout-strategy: PING # 心跳超时策略，支持发送 PING 和 CLOSE 断开连接，默认：最大努力发送 PING。（2.4.3 开始支持）
clean-start: true # session 保留 2.5.x 使用 clean-start，老版本用 clean-session，默认：true
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 密码
topic1: /test/#

183
example/mica-mqtt-example/pom.xml
View File

@@ -1,183 +0,0 @@
<?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>
<artifactId>mica-mqtt-example</artifactId>
<name>${project.artifactId}</name>
<parent>
<groupId>org.dromara.mica-mqtt</groupId>
<artifactId>example</artifactId>
<version>${revision}</version>
</parent>
<properties>
<graalvm.version>25.0.1</graalvm.version>
<mainClass.server>org.dromara.mica.mqtt.server.MqttServerTest</mainClass.server>
<mainClass.client>org.dromara.mica.mqtt.client.MqttClientTest</mainClass.client>
</properties>
<dependencies>
<dependency>
<groupId>org.dromara.mica-mqtt</groupId>
<artifactId>mica-mqtt-client</artifactId>
</dependency>
<dependency>
<groupId>org.dromara.mica-mqtt</groupId>
<artifactId>mica-mqtt-server</artifactId>
</dependency>
<dependency>
<groupId>com.google.code.gson</groupId>
<artifactId>gson</artifactId>
</dependency>
</dependencies>
<build>
<finalName>${project.artifactId}</finalName>
</build>
<profiles>
<profile>
<id>jar</id>
<activation>
<activeByDefault>true</activeByDefault>
</activation>
<dependencies>
<!-- 非 GraalVM 环境用 tinylog -->
<dependency>
<groupId>org.tinylog</groupId>
<artifactId>slf4j-tinylog</artifactId>
</dependency>
<dependency>
<groupId>org.tinylog</groupId>
<artifactId>tinylog-impl</artifactId>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-jar-plugin</artifactId>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-assembly-plugin</artifactId>
<version>3.8.0</version>
<configuration>
<archive>
<manifest>
<mainClass>${mainClass.server}</mainClass>
</manifest>
</archive>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
<appendAssemblyId>false</appendAssemblyId>
</configuration>
<executions>
<execution>
<id>make-assembly</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</profile>
<profile>
<id>mqtt-server-graal</id>
<dependencies>
<!-- GraalVM 环境使用 jdk log -->
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-jdk14</artifactId>
</dependency>
<!-- GraalVM -->
<dependency>
<groupId>org.graalvm.sdk</groupId>
<artifactId>graal-sdk</artifactId>
<version>${graalvm.version}</version>
<scope>provided</scope>
</dependency>
</dependencies>
<build>
<finalName>mqtt-server-graal</finalName>
<plugins>
<plugin>
<groupId>org.graalvm.nativeimage</groupId>
<artifactId>native-image-maven-plugin</artifactId>
<version>21.2.0</version>
<executions>
<execution>
<goals>
<goal>native-image</goal>
</goals>
<phase>package</phase>
</execution>
</executions>
<configuration>
<skip>false</skip>
<imageName>mqtt-server-graalvm</imageName>
<mainClass>${mainClass.server}</mainClass>
<buildArgs>
-H:+RemoveSaturatedTypeFlows
--allow-incomplete-classpath
--no-fallback
</buildArgs>
</configuration>
</plugin>
</plugins>
</build>
</profile>
<profile>
<id>mqtt-client-graal</id>
<dependencies>
<!-- GraalVM 环境使用 jdk log -->
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-jdk14</artifactId>
</dependency>
<!-- GraalVM -->
<dependency>
<groupId>org.graalvm.sdk</groupId>
<artifactId>graal-sdk</artifactId>
<version>${graalvm.version}</version>
<scope>provided</scope>
</dependency>
</dependencies>
<build>
<finalName>mqtt-client-graal</finalName>
<plugins>
<plugin>
<groupId>org.graalvm.nativeimage</groupId>
<artifactId>native-image-maven-plugin</artifactId>
<version>21.2.0</version>
<executions>
<execution>
<goals>
<goal>native-image</goal>
</goals>
<phase>package</phase>
</execution>
</executions>
<configuration>
<skip>false</skip>
<imageName>mqtt-client-graalvm</imageName>
<mainClass>${mainClass.client}</mainClass>
<buildArgs>
-H:+RemoveSaturatedTypeFlows
--allow-incomplete-classpath
--no-fallback
</buildArgs>
</configuration>
</plugin>
</plugins>
</build>
</profile>
</profiles>
</project>

65
example/mica-mqtt-example/src/main/java/org/dromara/mica/mqtt/aliyun/MqttClientTest.java
View File

@@ -1,65 +0,0 @@
/*
* Copyright (c) 2019-2029, Dreamlu 卢春梦 (596392912@qq.com & www.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 org.dromara.mica.mqtt.aliyun;
import org.dromara.mica.mqtt.core.client.MqttClient;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ThreadLocalRandom;
/**
* 客户端测试
*
* @author L.cm
*/
public class MqttClientTest {
public static void main(String[] args) {
String productKey = "g27jB42P9hm";
String deviceName = "3dbc1cb4";
String deviceSecret = "";
// 计算MQTT连接参数。
MqttSign sign = new MqttSign(productKey, deviceName, deviceSecret);
String username = sign.getUsername();
String password = sign.getPassword();
String clientId = sign.getClientId();
System.out.println("username: " + username);
System.out.println("password: " + password);
System.out.println("clientid: " + clientId);
// 初始化 mqtt 客户端
MqttClient client = MqttClient.create()
.ip(productKey + ".iot-as-mqtt.cn-shanghai.aliyuncs.com")
.port(443)
.username(username)
.password(password)
.clientId(clientId)
.connectSync();
client.subQos0("/sys/" + productKey + '/' + deviceName + "/thing/event/property/post_reply", (context, topic, message, payload) -> {
System.out.println(topic + '\t' + new String(payload, StandardCharsets.UTF_8));
});
client.schedule(() -> {
int LightSwitch = ThreadLocalRandom.current().nextBoolean() ? 0 : 1;
String content = "{\"id\":\"1\",\"version\":\"1.0\",\"params\":{\"LightSwitch\":" + LightSwitch + "}}";
client.publish("/sys/" + productKey + "/" + deviceName + "/thing/event/property/post", content.getBytes(StandardCharsets.UTF_8));
}, 3000);
}
}

81
example/mica-mqtt-example/src/main/java/org/dromara/mica/mqtt/aliyun/MqttSign.java
View File

@@ -1,81 +0,0 @@
/*
* Copyright (c) 2019-2029, Dreamlu 卢春梦 (596392912@qq.com & www.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 org.dromara.mica.mqtt.aliyun;
import org.tio.utils.mica.DigestUtils;
import java.util.Objects;
/**
* 阿里云 mqtt 签名方式
*
* @author L.cm
*/
public class MqttSign {
/**
* 用户名
*/
private final String username;
/**
* 密码
*/
private final String password;
/**
* 客户端id
*/
private final String clientId;
public MqttSign(String productKey, String deviceName, String deviceSecret) {
Objects.requireNonNull(productKey, "productKey is null");
Objects.requireNonNull(deviceName, "deviceName is null");
Objects.requireNonNull(deviceSecret, "deviceSecret is null");
this.username = deviceName + '&' + productKey;
String timestamp = Long.toString(System.currentTimeMillis());
this.password = getPassword(productKey, deviceName, deviceSecret, timestamp);
this.clientId = getClientId(productKey, deviceName, timestamp);
}
public String getUsername() {
return username;
}
public String getPassword() {
return password;
}
public String getClientId() {
return clientId;
}
private static String getPassword(String productKey, String deviceName, String deviceSecret, String timestamp) {
String plainPwd = "clientId" + productKey + '.' + deviceName + "deviceName" +
deviceName + "productKey" + productKey + "timestamp" + timestamp;
return hmacSha256(plainPwd, deviceSecret);
}
private static String getClientId(String productKey, String deviceName, String timestamp) {
return productKey + '.' + deviceName + "|timestamp=" + timestamp + ",_v=paho-java-1.0.0,securemode=2,signmethod=hmacsha256|";
}
private static String hmacSha256(String plainText, String key) {
if (plainText == null || key == null) {
return null;
}
return DigestUtils.hmacSha256Hex(plainText, key);
}
}

72
example/mica-mqtt-example/src/main/java/org/dromara/mica/mqtt/benchmark/MqttBenchmark.java
View File

@@ -1,72 +0,0 @@
/*
* Copyright (c) 2019-2029, Dreamlu 卢春梦 (596392912@qq.com & www.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 org.dromara.mica.mqtt.benchmark;
import org.dromara.mica.mqtt.core.client.MqttClient;
import org.tio.utils.hutool.StrUtil;
import org.tio.utils.thread.ThreadUtils;
import org.tio.utils.thread.pool.SynThreadPoolExecutor;
import org.tio.utils.timer.DefaultTimerTaskService;
import org.tio.utils.timer.TimerTaskService;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
/**
* mqtt 压力测试
*
* @author L.cm
*/
public class MqttBenchmark {
public static void main(String[] args) {
// 注意： windows 上需要修改最大的 Tcp 连接数，不然超不过 2W。
// 《修改Windows服务器最大的Tcp连接数》：https://www.jianshu.com/p/00136a97d2d8
int connCount = 5_0000;
String ip = "127.0.0.1";
// 优化：使用ArrayList+预分配容量，性能比CopyOnWriteArrayList好很多
// 对于压测场景，不需要线程安全的写操作（因为是单线程创建客户端）
final List<MqttClient> clientList = new ArrayList<>(connCount);
SynThreadPoolExecutor tioExecutor = ThreadUtils.getTioExecutor();
ExecutorService groupExecutor = ThreadUtils.getGroupExecutor();
// 自定义全局 taskService，避免每个 client new，创建过多线程
TimerTaskService taskService = new DefaultTimerTaskService(200L, 60);
for (int i = 0; i < connCount; i++) {
newClient(ip, i, clientList, tioExecutor, groupExecutor, taskService);
}
}
private static void newClient(String ip, int i, final List<MqttClient> clientList,
SynThreadPoolExecutor tioExecutor,
ExecutorService groupExecutor,
TimerTaskService taskService) {
MqttClient client = MqttClient.create()
.ip(ip)
.clientId(StrUtil.getNanoId() + i)
.readBufferSize(128)
// 取消自动重连
.reconnect(false)
.tioExecutor(tioExecutor)
.groupExecutor(groupExecutor)
.mqttExecutor(tioExecutor)
.taskService(taskService)
.connect();
clientList.add(client);
}
}

59
example/mica-mqtt-example/src/main/java/org/dromara/mica/mqtt/benchmark/MqttPublishBench.java
View File

@@ -1,59 +0,0 @@
package org.dromara.mica.mqtt.benchmark;
import org.dromara.mica.mqtt.codec.MqttQoS;
import org.dromara.mica.mqtt.core.client.MqttClient;
import org.tio.utils.hutool.StrUtil;
import org.tio.utils.thread.ThreadUtils;
import org.tio.utils.thread.pool.SynThreadPoolExecutor;
import org.tio.utils.timer.DefaultTimerTaskService;
import org.tio.utils.timer.TimerTaskService;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
* mqtt 发布端测试
*
* @author L.cm
*/
public class MqttPublishBench {
public static void main(String[] args) {
int clientCount = 10;
int publishCount = 10000;
MqttQoS qos = MqttQoS.QOS0;
List<MqttClient> clients = getClient(clientCount);
Executors.newScheduledThreadPool(ThreadUtils.AVAILABLE_PROCESSORS).scheduleWithFixedDelay(() -> {
for (MqttClient mqttClient : clients) {
for (int j = 0; j < publishCount; j++) {
byte[] payload = new byte[1024 + j];
Arrays.fill(payload, (byte) -1);
mqttClient.publish("/topic/" + j, payload, qos);
}
}
}, 1L, 1L, TimeUnit.SECONDS);
}
public static List<MqttClient> getClient(int clientCount) {
SynThreadPoolExecutor tioExecutor = ThreadUtils.getTioExecutor();
ExecutorService groupExecutor = ThreadUtils.getGroupExecutor();
TimerTaskService taskService = new DefaultTimerTaskService();
List<MqttClient> clients = new ArrayList<>();
for (int i = 0; i < clientCount; i++) {
MqttClient client = MqttClient.create()
.clientId(StrUtil.getNanoId())
.tioExecutor(tioExecutor)
.groupExecutor(groupExecutor)
.mqttExecutor(groupExecutor)
.taskService(taskService)
.connectSync();
clients.add(client);
}
return clients;
}
}

38
example/mica-mqtt-example/src/main/java/org/dromara/mica/mqtt/benchmark/MqttServerBench.java
View File

@@ -1,38 +0,0 @@
/*
* 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 org.dromara.mica.mqtt.benchmark;
import org.dromara.mica.mqtt.core.server.MqttServer;
/**
* mqtt 服务端测试
*
* @author L.cm
*/
public class MqttServerBench {
public static void main(String[] args) {
// 设定日志级别为 error
System.setProperty("tinylog.writer.level", "error");
// 启动 mqtt 服务
MqttServer.create()
.enableMqtt()
.statEnable(false)
.start();
}
}

47
example/mica-mqtt-example/src/main/java/org/dromara/mica/mqtt/broker/DeviceA.java
View File

@@ -1,47 +0,0 @@
/*
* Copyright (c) 2019-2029, Dreamlu 卢春梦 (596392912@qq.com & www.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 org.dromara.mica.mqtt.broker;
import org.dromara.mica.mqtt.core.client.MqttClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.nio.charset.StandardCharsets;
/**
* 设备 A，这里默认 APP 应用端
*
* @author L.cm
*/
public class DeviceA {
private static final Logger logger = LoggerFactory.getLogger(DeviceA.class);
public static void main(String[] args) {
// 初始化 mqtt 客户端
MqttClient client = MqttClient.create()
.ip("127.0.0.1")
.port(1883)
.username("admin")
.password("123456")
.connectSync();
client.subQos0("/a/door/open", (context, topic, message, payload) -> {
logger.info(topic + '\t' + new String(payload, StandardCharsets.UTF_8));
});
}
}

46
example/mica-mqtt-example/src/main/java/org/dromara/mica/mqtt/broker/DeviceB.java
View File

@@ -1,46 +0,0 @@
/*
* Copyright (c) 2019-2029, Dreamlu 卢春梦 (596392912@qq.com & www.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 org.dromara.mica.mqtt.broker;
import org.dromara.mica.mqtt.core.client.MqttClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.nio.charset.StandardCharsets;
/**
* 设备 B，这里默认 web 端
*
* @author L.cm
*/
public class DeviceB {
private static final Logger logger = LoggerFactory.getLogger(DeviceB.class);
public static void main(String[] args) {
// 初始化 mqtt 客户端
MqttClient client = MqttClient.create()
.ip("127.0.0.1")
.port(1883)
.username("admin")
.password("123456")
.connectSync();
client.subQos0("/a/door/open", (context, topic, message, payload) -> {
logger.info(topic + '\t' + new String(payload, StandardCharsets.UTF_8));
});
}
}

43
example/mica-mqtt-example/src/main/java/org/dromara/mica/mqtt/broker/DeviceC.java
View File

@@ -1,43 +0,0 @@
/*
* Copyright (c) 2019-2029, Dreamlu 卢春梦 (596392912@qq.com & www.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 org.dromara.mica.mqtt.broker;
import org.dromara.mica.mqtt.core.client.MqttClient;
/**
* 设备 C，每 5 秒上报一个数据
*
* @author L.cm
*/
public class DeviceC {
public static void main(String[] args) {
// 初始化 mqtt 客户端
MqttClient client = MqttClient.create()
.ip("127.0.0.1")
.port(1883)
.username("admin")
.password("123456")
.connectSync();
client.schedule(() -> {
client.publish("/a/door/open", null);
}, 5000);
}
}

36
example/mica-mqtt-example/src/main/java/org/dromara/mica/mqtt/broker/DeviceD.java
View File

@@ -1,36 +0,0 @@
package org.dromara.mica.mqtt.broker;
import org.dromara.mica.mqtt.codec.MqttQoS;
import org.dromara.mica.mqtt.core.annotation.MqttRetain;
import org.dromara.mica.mqtt.core.client.MqttClient;
import org.dromara.mica.mqtt.core.annotation.MqttClientPublish;
import org.dromara.mica.mqtt.core.annotation.MqttPayload;
/**
* @author ChangJin Wei (魏昌进)
*/
public class DeviceD {
public static void main(String[] args) {
// 初始化 mqtt 客户端
MqttClient client = MqttClient.create()
.ip("127.0.0.1")
.port(1883)
.username("admin")
.password("123456")
.connectSync();
DoorClient doorClient = client.getInterface(DoorClient.class);
client.schedule(() -> {
doorClient.sendMessage("open", false);
}, 1000);
}
public interface DoorClient {
@MqttClientPublish(value = "/a/door/open", qos = MqttQoS.QOS0)
void sendMessage(@MqttPayload String message, @MqttRetain boolean retain);
}
}

42
example/mica-mqtt-example/src/main/java/org/dromara/mica/mqtt/broker/Server.java
View File

@@ -1,42 +0,0 @@
/*
* Copyright (c) 2019-2029, Dreamlu 卢春梦 (596392912@qq.com & www.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 org.dromara.mica.mqtt.broker;
import org.dromara.mica.mqtt.core.server.MqttServer;
/**
* 服务端，单纯的做消息转发
*
* @author L.cm
*/
public class Server {
/**
* 客户端 A 模拟 APP 端订阅 `/a/door/open`，
* 客户端 B 模拟 web 网页端 mqtt.js 订阅 `/a/door/open`，
* Mqtt 服务端实现 `IMqttMessageListener`，将消息转交给 `AbstractMqttMessageDispatcher`（自定义实现）处理。
* 客户端 C 定时上报转态给 `/a/door/open`
* 结果：A 和 B 将收到 C 或 D 发布的消息，并完成相应的效果展示。
*/
public static void main(String[] args) {
// 启动服务，mica-mqtt 1.3.x 已经默认为 broker 模式
MqttServer.create()
.enableMqtt()
.debug()
.start();
}
}

79
example/mica-mqtt-example/src/main/java/org/dromara/mica/mqtt/client/Mqtt5ClientTest.java
View File

@@ -1,79 +0,0 @@
/*
* 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 org.dromara.mica.mqtt.client;
import org.dromara.mica.mqtt.codec.message.MqttPublishMessage;
import org.dromara.mica.mqtt.codec.MqttQoS;
import org.dromara.mica.mqtt.codec.MqttVersion;
import org.dromara.mica.mqtt.core.client.IMqttClientMessageListener;
import org.dromara.mica.mqtt.core.client.MqttClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tio.core.ChannelContext;
import java.nio.charset.StandardCharsets;
/**
* 客户端测试
*
* @author L.cm
*/
public class Mqtt5ClientTest {
private static final Logger logger = LoggerFactory.getLogger(Mqtt5ClientTest.class);
public static void main(String[] args) {
// 初始化 mqtt 客户端
MqttClient client = MqttClient.create()
.ip("127.0.0.1")
.port(1883)
.username("mica")
.password("mica")
.version(MqttVersion.MQTT_5)
.cleanStart(false)
.sessionExpiryIntervalSecs(7200)
.connectListener(new MqttClientConnectListener())
.willMessage(builder -> {
builder.topic("/test/offline")
.messageText("down")
.retain(false)
.qos(MqttQoS.QOS0); // 遗嘱消息
})
// 同步连接，也可以使用 connect() 异步（可以避免 broker 没启动照成启动卡住），但是下面的订阅和发布可能还没连接成功。
.connectSync();
client.subQos0("/test/123", new IMqttClientMessageListener() {
@Override
public void onSubscribed(ChannelContext context, String topicFilter, MqttQoS mqttQoS) {
// 订阅成功之后触发，可在此处做一些业务逻辑
logger.info("topicFilter:{} MqttQoS:{} 订阅成功！！！", topicFilter, mqttQoS);
}
@Override
public void onMessage(ChannelContext context, String topic, MqttPublishMessage message, byte[] payload) {
logger.info(topic + '\t' + new String(payload, StandardCharsets.UTF_8));
}
});
client.publish("/test/client", "mica最牛皮1".getBytes(StandardCharsets.UTF_8));
client.publish("/test/client", "mica最牛皮2".getBytes(StandardCharsets.UTF_8));
client.publish("/test/client", "mica最牛皮3".getBytes(StandardCharsets.UTF_8));
client.schedule(() -> {
client.publish("/test/client", "mica最牛皮".getBytes(StandardCharsets.UTF_8));
}, 1000);
}
}

40
example/mica-mqtt-example/src/main/java/org/dromara/mica/mqtt/client/MqttClientConnTest.java
View File

@@ -1,40 +0,0 @@
/*
* 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 org.dromara.mica.mqtt.client;
import org.dromara.mica.mqtt.codec.codes.MqttConnectReasonCode;
import org.dromara.mica.mqtt.core.client.MqttClient;
/**
* 客户端测试
*
* @author L.cm
*/
public class MqttClientConnTest {
public static void main(String[] args) {
// 初始化 mqtt 客户端
MqttConnectReasonCode reasonCode = MqttClient.create()
// .ip("127.0.0.1")
.ip("mqtt.dreamlu.net")
.port(1883)
.username("mica")
.password("mica1")
.connectTest();
System.out.println(reasonCode);
}
}

48
example/mica-mqtt-example/src/main/java/org/dromara/mica/mqtt/client/MqttClientConnectListener.java
View File

@@ -1,48 +0,0 @@
/*
* 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 org.dromara.mica.mqtt.client;
import org.dromara.mica.mqtt.core.client.IMqttClientConnectListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tio.core.ChannelContext;
/**
* 客户端连接状态监听
*
* @author L.cm
*/
public class MqttClientConnectListener implements IMqttClientConnectListener {
private static final Logger logger = LoggerFactory.getLogger(MqttClientConnectListener.class);
@Override
public void onConnected(ChannelContext context, boolean isReconnect) {
String clientId = context.getId();
if (isReconnect) {
logger.info("重连 mqtt 服务器重连成功... clientId:{}", clientId);
} else {
logger.info("连接 mqtt 服务器成功... clientId:{}", clientId);
}
}
@Override
public void onDisconnect(ChannelContext context, Throwable throwable, String remark, boolean isRemove) {
String clientId = context.getId();
logger.error("mqtt 链接断开 remark:{} isRemove:{} clientId:{}", remark, isRemove, clientId, throwable);
}
}

50
example/mica-mqtt-example/src/main/java/org/dromara/mica/mqtt/client/MqttClientGlobalTest.java
View File

@@ -1,50 +0,0 @@
/*
* 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 org.dromara.mica.mqtt.client;
import org.dromara.mica.mqtt.core.client.MqttClient;
import org.tio.utils.buffer.ByteBufferUtil;
/**
* 客户端全局订阅测试
*
* @author L.cm
*/
public class MqttClientGlobalTest {
public static void main(String[] args) {
// 初始化 mqtt 客户端
MqttClient.create()
.ip("127.0.0.1")
.port(1883)
.username("admin")
.password("123456")
// 采用 globalSubscribe，保留 session 停机重启后，可以接受到离线消息，注意：clientId 要不能变化。
.clientId("globalTest")
.cleanStart(false)
// 全局订阅的 topic
.globalSubscribe("/test", "/test/123", "/debug/#")
// 全局监听，也会监听到服务端 http api 订阅的数据
.globalMessageListener((context, topic, message, payload) -> {
System.out.println("topic:\t" + topic);
System.out.println("payload:\t" + ByteBufferUtil.toString(payload));
})
// .debug()
.connectSync();
}
}

55
example/mica-mqtt-example/src/main/java/org/dromara/mica/mqtt/client/MqttClientSyncTest.java
View File

@@ -1,55 +0,0 @@
/*
* 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 org.dromara.mica.mqtt.client;
import org.dromara.mica.mqtt.core.client.MqttClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.nio.charset.StandardCharsets;
/**
* 客户端测试
*
* @author L.cm
*/
public class MqttClientSyncTest {
private static final Logger logger = LoggerFactory.getLogger(MqttClientSyncTest.class);
public static void main(String[] args) {
// 初始化 mqtt 客户端
MqttClient client = MqttClient.create()
.ip("127.0.0.1")
.port(1883)
.username("mica")
.password("mica")
.connectListener(new MqttClientConnectListener())
// 同步连接，注意：连接会阻塞
.connectSync();
client.subQos0("/test/#", (context, topic, message, payload) -> {
logger.info(topic + '\t' + new String(payload, StandardCharsets.UTF_8));
});
client.unSubscribe("/test/#", "/test/123");
// 连接上之后发送消息，注意：连接时出现异常等就不会发出
client.publish("/test/client", "mica最牛皮".getBytes(StandardCharsets.UTF_8));
// 2.3.0 开始支持，可停止
// client.stop();
}
}

129
example/mica-mqtt-example/src/main/java/org/dromara/mica/mqtt/client/MqttClientTest.java
View File

@@ -1,129 +0,0 @@
/*
* 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 org.dromara.mica.mqtt.client;
import org.dromara.mica.mqtt.codec.message.MqttPublishMessage;
import org.dromara.mica.mqtt.codec.MqttQoS;
import org.dromara.mica.mqtt.core.client.IMqttClientMessageListener;
import org.dromara.mica.mqtt.core.client.MqttClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tio.core.ChannelContext;
import java.nio.charset.StandardCharsets;
/**
* 客户端测试
*
* @author L.cm
*/
public class MqttClientTest {
private static final Logger logger = LoggerFactory.getLogger(MqttClientTest.class);
public static void main(String[] args) {
// 初始化 mqtt 客户端
MqttClient client = MqttClient.create()
.ip("127.0.0.1")
.port(1883)
.username("mica")
.password("mica")
// 绑定网卡 v2.5.1 添加，支持 bindIp（网卡ip）和 bindNetworkInterface（网卡名）取一即可
// .bindIp("127.0.0.1")
// .bindNetworkInterface("lo")
// 如果包体过大，建议将此参数设置和 maxBytesInMessage 一样大
// .readBufferSize(1024 * 10)
// 最大包体长度,如果包体过大需要设置此参数
// .maxBytesInMessage(1024 * 10)
// .version(MqttVersion.MQTT_5)
// 连接监听
.connectListener(new MqttClientConnectListener())
// 遗嘱消息
.willMessage(builder -> {
builder.topic("/test/offline")
.messageText("down")
.retain(false)
.qos(MqttQoS.QOS0)
// mqtt5 遗嘱消息属性
.willProperties(props -> {
props.setWillDelayInterval(1000);
props.setContentType("text/plain");
});
})
// 同步连接，也可以使用 connect() 异步（可以避免 broker 没启动照成启动卡住），但是下面的订阅和发布可能还没连接成功。
.connectSync();
client.subQos0("/test/#", new IMqttClientMessageListener() {
@Override
public void onSubscribed(ChannelContext context, String topicFilter, MqttQoS mqttQoS) {
// 订阅成功之后触发，可在此处做一些业务逻辑
logger.info("topicFilter:{} MqttQoS:{} 订阅成功！！！", topicFilter, mqttQoS);
}
@Override
public void onMessage(ChannelContext context, String topic, MqttPublishMessage message, byte[] payload) {
logger.info(topic + '\t' + new String(payload, StandardCharsets.UTF_8));
}
});
client.subQos0("/test/1", new IMqttClientMessageListener() {
@Override
public void onSubscribed(ChannelContext context, String topicFilter, MqttQoS mqttQoS) {
// 订阅成功之后触发，可在此处做一些业务逻辑
logger.info("topicFilter:{} MqttQoS:{} 订阅成功！！！", topicFilter, mqttQoS);
}
@Override
public void onMessage(ChannelContext context, String topic, MqttPublishMessage message, byte[] payload) {
logger.info(topic + '\t' + new String(payload, StandardCharsets.UTF_8));
}
});
client.subQos0("/test/2", new IMqttClientMessageListener() {
@Override
public void onSubscribed(ChannelContext context, String topicFilter, MqttQoS mqttQoS) {
// 订阅成功之后触发，可在此处做一些业务逻辑
logger.info("topicFilter:{} MqttQoS:{} 订阅成功！！！", topicFilter, mqttQoS);
}
@Override
public void onMessage(ChannelContext context, String topic, MqttPublishMessage message, byte[] payload) {
logger.info(topic + '\t' + new String(payload, StandardCharsets.UTF_8));
}
});
client.subQos0("/test/3", new IMqttClientMessageListener() {
@Override
public void onSubscribed(ChannelContext context, String topicFilter, MqttQoS mqttQoS) {
// 订阅成功之后触发，可在此处做一些业务逻辑
logger.info("topicFilter:{} MqttQoS:{} 订阅成功！！！", topicFilter, mqttQoS);
}
@Override
public void onMessage(ChannelContext context, String topic, MqttPublishMessage message, byte[] payload) {
logger.info(topic + '\t' + new String(payload, StandardCharsets.UTF_8));
}
});
client.publish("/test/client", "mica最牛皮1".getBytes(StandardCharsets.UTF_8));
client.publish("/test/client", "mica最牛皮2".getBytes(StandardCharsets.UTF_8));
client.publish("/test/client", "mica最牛皮3".getBytes(StandardCharsets.UTF_8));
client.schedule(() -> {
client.publish("/test/client", "mica最牛皮".getBytes(StandardCharsets.UTF_8));
}, 2000);
}
}

70
example/mica-mqtt-example/src/main/java/org/dromara/mica/mqtt/huawei/MqttClientTest.java
View File

@@ -1,70 +0,0 @@
/*
* Copyright (c) 2019-2029, Dreamlu 卢春梦 (596392912@qq.com & www.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 org.dromara.mica.mqtt.huawei;
import org.dromara.mica.mqtt.core.client.MqttClient;
import java.nio.charset.StandardCharsets;
/**
* 客户端测试
*
* @author L.cm
*/
public class MqttClientTest {
public static void main(String[] args) {
// 设备id和密钥，请从华为云iot获取
String deviceId = "630eb6f8664c6f7938db6ef0_test";
String deviceSecret = "";
// 计算MQTT连接参数。
MqttSign sign = new MqttSign(deviceId, deviceSecret);
String username = sign.getUsername();
String password = sign.getPassword();
String clientId = sign.getClientId();
System.out.println("username: " + username);
System.out.println("password: " + password);
System.out.println("clientid: " + clientId);
// 初始化 mqtt 客户端
MqttClient client = MqttClient.create()
.ip("iot-mqtts.cn-north-4.myhuaweicloud.com")
.port(8883)
.username(username)
.password(password)
.clientId(clientId)
.useSsl()
.connectSync();
// 订阅命令下发topic
String cmdRequestTopic = "$oc/devices/" + deviceId + "/sys/commands/#";
client.subQos0(cmdRequestTopic, (context, topic, message, payload) -> {
System.out.println(topic + '\t' + new String(payload, StandardCharsets.UTF_8));
});
// 属性上报消息
String reportTopic = "$oc/devices/" + deviceId + "/sys/properties/report";
String jsonMsg = "{\"services\":[{\"service_id\":\"Temperature\", \"properties\":{\"value\":57}},{\"service_id\":\"Battery\",\"properties\":{\"level\":88}}]}";
client.schedule(() -> {
client.publish(reportTopic, jsonMsg.getBytes(StandardCharsets.UTF_8));
}, 3000);
}
}

89
example/mica-mqtt-example/src/main/java/org/dromara/mica/mqtt/huawei/MqttSign.java
View File

@@ -1,89 +0,0 @@
/*
* Copyright (c) 2019-2029, Dreamlu 卢春梦 (596392912@qq.com & www.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 org.dromara.mica.mqtt.huawei;
import org.tio.utils.mica.DigestUtils;
import java.time.Instant;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Objects;
/**
* 华为云 mqtt 签名方式
*
* @author L.cm
*/
public class MqttSign {
/**
* 用户名
*/
private final String username;
/**
* 密码
*/
private final String password;
/**
* 客户端id
*/
private final String clientId;
public MqttSign(String deviceId, String deviceSecret) {
Objects.requireNonNull(deviceId, "deviceId is null");
Objects.requireNonNull(deviceSecret, "deviceSecret is null");
this.username = deviceId;
String timestamp = getTimeStamp();
this.password = getPassword(deviceSecret, timestamp);
this.clientId = getClientId(deviceId, timestamp);
}
public String getUsername() {
return username;
}
public String getPassword() {
return password;
}
public String getClientId() {
return clientId;
}
private static String getPassword(String deviceSecret, String timestamp) {
return hmacSha256(deviceSecret, timestamp);
}
private static String getClientId(String deviceId, String timestamp) {
return deviceId + "_0_0_" + timestamp;
}
/***
* 要求：10位数字
*/
private static String getTimeStamp() {
return ZonedDateTime.ofInstant(Instant.now(), ZoneId.of("UTC")).format(DateTimeFormatter.ofPattern("yyyyMMddHH"));
}
/***
* 调用sha256算法进行哈希
*/
private static String hmacSha256(String message, String tStamp) {
return DigestUtils.hmacSha256Hex(message, tStamp);
}
}

40
example/mica-mqtt-example/src/main/java/org/dromara/mica/mqtt/nginx/MqttServerProxyProtocol.java
View File

@@ -1,40 +0,0 @@
/*
* 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 org.dromara.mica.mqtt.nginx;
import org.dromara.mica.mqtt.core.server.MqttServer;
/**
* mqtt 服务端测试
*
* @author L.cm
*/
public class MqttServerProxyProtocol {
public static void main(String[] args) {
MqttServer.create()
.enableMqtt()
.enableMqttWs()
// 开启代理协议
.proxyProtocolEnable()
.statEnable(false)
.debug()
.start();
}
}

58
example/mica-mqtt-example/src/main/java/org/dromara/mica/mqtt/proxy/MqttClientProxy.java
View File

@@ -1,58 +0,0 @@
/*
* 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 org.dromara.mica.mqtt.proxy;
import org.dromara.mica.mqtt.codec.MqttQoS;
import org.dromara.mica.mqtt.core.client.MqttClient;
/**
* 2 个 mqtt 服务间，使用 2 个 client 做数据传输
*
* @author L.cm
*/
public class MqttClientProxy {
public static void main(String[] args) {
MqttClient client1 = MqttClient.create()
.ip("ip1")
.port(1883)
.clientId("clientI")
.username("mica")
.password("mica")
.debug()
.connectSync();
MqttClient client2 = MqttClient.create()
.ip("ip2")
.port(1883)
.clientId("client2")
.username("mica")
.password("mica")
.debug()
.connectSync();
String[] topics = new String[]{
"$share/test/link/product1/+/event/+/post",
"$share/test/link/product2/+/event/+/post",
"$share/test/link/product3/+/event/+/post"
};
client1.subscribe(topics, MqttQoS.QOS0, (context, topic, message, payload) -> {
client2.publish(topic, payload);
});
}
}

67
example/mica-mqtt-example/src/main/java/org/dromara/mica/mqtt/proxy/MqttServerProxy.java
View File

@@ -1,67 +0,0 @@
/*
* 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 org.dromara.mica.mqtt.proxy;
import org.dromara.mica.mqtt.core.client.MqttClient;
import org.dromara.mica.mqtt.core.server.MqttServer;
import org.dromara.mica.mqtt.server.MqttConnectStatusListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tio.utils.buffer.ByteBufferUtil;
/**
* mqtt 服务端代理到另外一个服务端
*
* @author L.cm
*/
public class MqttServerProxy {
private static final Logger logger = LoggerFactory.getLogger(MqttServerProxy.class);
public static void main(String[] args) {
// 需要将数据发往的服务端
MqttClient client = MqttClient.create()
.ip("ip")
.port(1883)
.clientId("proxy")
.username("mica")
.password("mcia")
.debug()
.connectSync();
// 接受数据的服务端
MqttServer.create()
.messageListener((context, clientId, topic, qoS, message) -> {
byte[] payload = message.payload();
logger.info("clientId:{} topic:{} payload:\n{}", clientId, topic, ByteBufferUtil.toString(payload));
// 转发数据
client.publish(topic, payload);
})
// 开启 mqtt tcp 协议
.enableMqtt()
// 开启 mqtt websocket
.enableMqttWs()
// 开启 http api 接口
.enableMqttHttpApi()
// 客户端连接状态监听
.connectStatusListener(new MqttConnectStatusListener())
// 开始 stat 监控
.statEnable()
// 开启 debug 信息日志
.debug()
.start();
}
}

42
example/mica-mqtt-example/src/main/java/org/dromara/mica/mqtt/server/MqttConnectStatusListener.java
View File

@@ -1,42 +0,0 @@
/*
* 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 org.dromara.mica.mqtt.server;
import org.dromara.mica.mqtt.core.server.event.IMqttConnectStatusListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tio.core.ChannelContext;
/**
* mqtt 连接状态
*
* @author L.cm
*/
public class MqttConnectStatusListener implements IMqttConnectStatusListener {
private static final Logger logger = LoggerFactory.getLogger(MqttConnectStatusListener.class);
@Override
public void online(ChannelContext context, String clientId, String username) {
logger.info("Mqtt clientId:{} username:{} online.", clientId, username);
}
@Override
public void offline(ChannelContext context, String clientId, String username, String reason) {
logger.info("Mqtt clientId:{} username:{} offline reason:{}.", clientId, username, reason);
}
}

58
example/mica-mqtt-example/src/main/java/org/dromara/mica/mqtt/server/MqttMessageInterceptor.java
View File

@@ -1,58 +0,0 @@
/*
* 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 org.dromara.mica.mqtt.server;
import org.dromara.mica.mqtt.codec.message.MqttMessage;
import org.dromara.mica.mqtt.core.server.interceptor.IMqttMessageInterceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tio.core.ChannelContext;
import org.tio.core.Node;
/**
* mqtt 消息拦截器
*
* @author L.cm
*/
public class MqttMessageInterceptor implements IMqttMessageInterceptor {
private static final Logger logger = LoggerFactory.getLogger(MqttMessageInterceptor.class);
@Override
public void onAfterReceivedBytes(ChannelContext context, int receivedBytes) throws Exception {
// 注意：此时 clientId 可能为空
String clientId = context.getBsId();
Node clientNode = context.getClientNode();
// ChannelStat channelStat = context.stat;
// 自定义规则，超限是可用 Tio.remove(context, "xxx超限"); 断开连接。
logger.info("===接收 client:{} clientId:{} data:{}b", clientNode, clientId, receivedBytes);
}
@Override
public void onAfterDecoded(ChannelContext context, MqttMessage message, int packetSize) {
// 注意：此时 clientId 可能为空
String clientId = context.getBsId();
Node clientNode = context.getClientNode();
logger.info("===解码 client:{} clientId:{} message:{}", clientNode, clientId, message);
}
@Override
public void onAfterHandled(ChannelContext context, MqttMessage message, long cost) throws Exception {
String clientId = context.getBsId();
Node clientNode = context.getClientNode();
logger.info("===处理完成 ip:{} clientId:{} message:{} 耗时:{}", clientNode, clientId, message, cost);
}
}

74
example/mica-mqtt-example/src/main/java/org/dromara/mica/mqtt/server/MqttServerTest.java
View File

@@ -1,74 +0,0 @@
/*
* 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 org.dromara.mica.mqtt.server;
import org.dromara.mica.mqtt.core.server.MqttServer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.nio.charset.StandardCharsets;
/**
* mqtt 服务端测试
*
* @author L.cm
*/
public class MqttServerTest {
private static final Logger logger = LoggerFactory.getLogger(MqttServerTest.class);
public static void main(String[] args) {
// 注意：为了能接受更多链接（降低内存），请添加 jvm 参数 -Xss129k
MqttServer mqttServer = MqttServer.create()
// 服务端 ip 默认为空，0.0.0.0，建议不要设置，端口 默认：1883
.enableMqtt(1883)
// 默认为： 8192（mqtt 默认最大消息大小），为了降低内存可以减小小此参数，如果消息过大 t-io 会尝试解析多次（建议根据实际业务情况而定）
.readBufferSize(8192)
// 最大包体长度
// .maxBytesInMessage(1024 * 100)
// mqtt 3.1 协议会校验 clientId 长度。
// .maxClientIdLength(64)
.messageListener((context, clientId, topic, qos, message) -> {
logger.info("clientId:{} payload:{}", clientId, new String(message.payload(), StandardCharsets.UTF_8));
})
// 客户端连接状态监听
.connectStatusListener(new MqttConnectStatusListener())
// 自定义消息拦截器
.addInterceptor(new MqttMessageInterceptor())
// 开启 websocket
.enableMqttWs()
// 开启 mqtt http 接口
.enableMqttHttpApi(builder ->
builder
.basicAuth("mica", "mica") // http basic 认证
.mcpServer() // 开启 mcp 服务
.build()
)
// 开始 stat 监控
.statEnable()
// 开启 debug 信息日志
.debug()
.start();
mqttServer.schedule(() -> {
String message = "mica最牛皮 " + System.currentTimeMillis();
mqttServer.publishAll("/test/123", message.getBytes(StandardCharsets.UTF_8));
}, 2000);
// 2.3.2 开始支持 stop 关闭
// mqttServer.stop();
}
}

54
example/mica-mqtt-example/src/main/java/org/dromara/mica/mqtt/ssl/SslMqttClientTest.java
View File

@@ -1,54 +0,0 @@
/*
* 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 org.dromara.mica.mqtt.ssl;
import org.dromara.mica.mqtt.core.client.MqttClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tio.utils.buffer.ByteBufferUtil;
import java.nio.charset.StandardCharsets;
/**
* 客户端测试
*
* @author L.cm
*/
public class SslMqttClientTest {
private static final Logger logger = LoggerFactory.getLogger(SslMqttClientTest.class);
public static void main(String[] args) {
// 初始化 mqtt 客户端
MqttClient client = MqttClient.create()
.ip("127.0.0.1")
.port(8883)
.username("mica")
.password("mica")
.useSsl("classpath:ssl/dreamlu.net.jks", "123456")
.connectSync();
client.subQos0("/test/#", (context, topic, message, payload) -> {
logger.info(topic + '\t' + ByteBufferUtil.toString(payload));
});
// 定时发送数据
client.schedule(() -> {
String message = "mica最牛皮 " + System.currentTimeMillis();
client.publish("/test/123", message.getBytes(StandardCharsets.UTF_8));
}, 5000);
}
}

54
example/mica-mqtt-example/src/main/java/org/dromara/mica/mqtt/ssl/SslMqttServerTest.java
View File

@@ -1,54 +0,0 @@
/*
* 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 org.dromara.mica.mqtt.ssl;
import org.dromara.mica.mqtt.core.server.MqttServer;
import org.dromara.mica.mqtt.server.MqttConnectStatusListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tio.utils.buffer.ByteBufferUtil;
import java.nio.charset.StandardCharsets;
/**
* mqtt 服务端测试
*
* @author L.cm
*/
public class SslMqttServerTest {
private static final Logger logger = LoggerFactory.getLogger(SslMqttServerTest.class);
public static void main(String[] args) {
MqttServer mqttServer = MqttServer.create()
.enableMqttSsl(builder ->
builder.useSsl("classpath:ssl/dreamlu.net.jks", "123456")
.build()
)
.messageListener((context, clientId, topic, qoS, message) -> {
logger.info("clientId:{} message:{} payload:{}", clientId, message, ByteBufferUtil.toString(message.payload()));
})
.connectStatusListener(new MqttConnectStatusListener())
.debug()
.start();
// 定时发送数据
mqttServer.schedule(() -> {
String message = "mica最牛皮 " + System.currentTimeMillis();
mqttServer.publishAll("/test/123", message.getBytes(StandardCharsets.UTF_8));
}, 5000);
}
}

1
example/mica-mqtt-example/src/main/resources/ssl/README.md
View File

@@ -1 +0,0 @@
ssl 自签双向证书详见：https://gitee.com/596392912/mica-mqtt/issues/I45GO7

16
example/mica-mqtt-example/src/main/resources/ssl/ca-cert.pem
View File

@@ -1,16 +0,0 @@
-----BEGIN CERTIFICATE-----
MIICejCCAeMCFGkzqgl8ilICj+PRCwIiUy3dPNuWMA0GCSqGSIb3DQEBCwUAMHwx
CzAJBgNVBAYTAkNOMQswCQYDVQQIDAJITjELMAkGA1UEBwwCQ1MxDTALBgNVBAoM
BFJNSlMxDTALBgNVBAsMBFJNSlMxFDASBgNVBAMMC2RyZWFtbHUubmV0MR8wHQYJ
KoZIhvcNAQkBFhA1OTYzOTI5MTJAcXEuY29tMB4XDTIyMDkwMzA4MjExNVoXDTMy
MDgzMTA4MjExNVowfDELMAkGA1UEBhMCQ04xCzAJBgNVBAgMAkhOMQswCQYDVQQH
DAJDUzENMAsGA1UECgwEUk1KUzENMAsGA1UECwwEUk1KUzEUMBIGA1UEAwwLZHJl
YW1sdS5uZXQxHzAdBgkqhkiG9w0BCQEWEDU5NjM5MjkxMkBxcS5jb20wgZ8wDQYJ
KoZIhvcNAQEBBQADgY0AMIGJAoGBAJxanDadRHd+D9jH5/mq7Pn3Fl915CiOaaJr
QS0ZxCqRtTaS9+JkFQJ4TnAezlKx/xDZu7G9pk7CJ6w+JQwfI+AwsAkOlrynyFbe
Hc9s6DZyZHXxkpgeQUqpnrkXkWG+jbh2aulWB4smQE/vPnpkjVGEe86+/JsYW3Sm
rFhL2xddAgMBAAEwDQYJKoZIhvcNAQELBQADgYEAP39PM0XpNzg3Bne63oXHRWDJ
bafiCwloO/SaxH7JtCBj1I05W3owwHSqWYadK+lg//tKf6TL+94GtW8s2VtpLGu7
Y5R2aakhywzbWVfrEK+kyvTG/4nP9tvcKwh8Iqr2XlllDLfsQeyLacb4+pcfDY3G
8nrbe0lBKepae8D0SvQ=
-----END CERTIFICATE-----

16
example/mica-mqtt-example/src/main/resources/ssl/client-cert.pem
View File

@@ -1,16 +0,0 @@
-----BEGIN CERTIFICATE-----
MIICejCCAeMCFBebamYN4nfF0CrZWuN68SXxTH/GMA0GCSqGSIb3DQEBCwUAMHwx
CzAJBgNVBAYTAkNOMQswCQYDVQQIDAJITjELMAkGA1UEBwwCQ1MxDTALBgNVBAoM
BFJNSlMxDTALBgNVBAsMBFJNSlMxFDASBgNVBAMMC2RyZWFtbHUubmV0MR8wHQYJ
KoZIhvcNAQkBFhA1OTYzOTI5MTJAcXEuY29tMB4XDTIyMDkwMzA4MjM1N1oXDTMy
MDgzMTA4MjM1N1owfDELMAkGA1UEBhMCQ04xCzAJBgNVBAgMAkhOMQswCQYDVQQH
DAJDUzENMAsGA1UECgwEUk1KUzENMAsGA1UECwwEUk1KUzEUMBIGA1UEAwwLZHJl
YW1sdS5uZXQxHzAdBgkqhkiG9w0BCQEWEDU5NjM5MjkxMkBxcS5jb20wgZ8wDQYJ
KoZIhvcNAQEBBQADgY0AMIGJAoGBAL3hH0vRYCsqs79ghUPskgr+bQyVkO/6U1H1
4yHJWyWeXT2tl+VTvXhou0Zk3yYtd6O3l241PgmAjwBDPyfmM2UU/zluXQzELEuf
AnC4Q2wl0vpoj7mGQeMuRpjZN3FblS0bRifvctx3ubWVvSzCFIaKLZgdrnyWm5Lm
kJQjRaEDAgMBAAEwDQYJKoZIhvcNAQELBQADgYEAkm/Mv9nEGE4xy3L4NmwgyZMo
j5CuarbCEv/MgXST8B/dJvtC7PbCzrb9BbUkiIUjCOnRC+2U0DbSAkHy7CCwHTLx
g9Syn1SMWn9SF74l74qVtuaRjMKCGwSlqBLH5/t1FdPoevY2a+zi5dAqwIhkjY84
uYOgxODe+kFBUbT6W3A=
-----END CERTIFICATE-----

15
example/mica-mqtt-example/src/main/resources/ssl/client-key.pem
View File

@@ -1,15 +0,0 @@
-----BEGIN RSA PRIVATE KEY-----
MIICXQIBAAKBgQC94R9L0WArKrO/YIVD7JIK/m0MlZDv+lNR9eMhyVslnl09rZfl
U714aLtGZN8mLXejt5duNT4JgI8AQz8n5jNlFP85bl0MxCxLnwJwuENsJdL6aI+5
hkHjLkaY2TdxW5UtG0Yn73Lcd7m1lb0swhSGii2YHa58lpuS5pCUI0WhAwIDAQAB
AoGAR/T8azsZesJgA/KMDkWkws3QfahgmNEAqlrIjJFGHWd6ZllW6u1lLDBkaDTp
7AnnAQAePwGmVOuHRc42LOSsLMX/D2wYsTGjLTT1w+fEDkQCVnDKV8ZYWA5fN7Zh
m5cLB5IB23L/Xvs8UMYQ8qWufv6BxVPr+cawtXOK3O91AoECQQD5PwGSrMpsNkVG
Ox1t1A9wVgelbq+9qANPl8TvdaZKvNsbWwI0PRqK7D1SJXJiMX4m/hc8YkIXJWsO
Wd1yLg6TAkEAwwZLncJYlYOp0v8PXk75rDR7KV9fdBPPOdnZ1XU/yFW3anwOt0OX
LvGL3X0A2f/tqOX9v+3LENCICF4gAv850QJBALo7Yal+giEoy8oWEX8mnAKLxVrO
wXEsQI0QEY36ki31vqFJ9vOhVFvI+GiQok7MPD5WTHZJ1KgGxV8LtnLCBxECQCPD
WcZ6RyhT1qaco0LWFK7hiNxTYvu0TkH7kxizwZiJL3NVgJVWzbiMDuv06l0Ps5NP
abLydlSFCQ0PxasHBqECQQC0lhmqkNbaN3lJyGDoEWm6Kb9z3eh3+9Fk4j328aYW
gQcBeRhKU8kdDTg2flOWS3sxrEysJYv8i9DPbX9RsRFd
-----END RSA PRIVATE KEY-----

BIN
example/mica-mqtt-example/src/main/resources/ssl/dreamlu.net.jks
View File

Binary file not shown.

8
example/mica-mqtt-example/src/main/resources/tinylog.properties
View File

@@ -1,8 +0,0 @@
writer = console
writer.format = {date: HH:mm:ss.SSS} [{thread}] {level} {class-name}.{method} : {message}
writer.level = info
# level
level@org.tio = warn
level@org.tio.client.TioClient = off
level@org.tio.server = info
level@org.dromara.mica.mqtt = info

1
example/mica-mqtt-example/src/test/java/net/dreamlu/iot/package-info.java
View File

@@ -1 +0,0 @@
package net.dreamlu.iot;

0
example/mica-mqtt-example/src/test/resources/test.yml
View File

49
example/mica-mqtt-server-solon-plugin-example/pom.xml
View File

@@ -1,49 +0,0 @@
<?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.dromara.mica-mqtt</groupId>
<artifactId>example</artifactId>
<version>${revision}</version>
</parent>
<artifactId>mica-mqtt-server-solon-plugin-example</artifactId>
<dependencies>
<dependency>
<groupId>org.noear</groupId>
<artifactId>solon-web</artifactId>
</dependency>
<dependency>
<groupId>org.dromara.mica-mqtt</groupId>
<artifactId>mica-mqtt-server-solon-plugin</artifactId>
</dependency>
<!-- 简单的本地定时任务调度 -->
<dependency>
<groupId>org.noear</groupId>
<artifactId>solon-scheduling-simple</artifactId>
</dependency>
<dependency>
<groupId>org.noear</groupId>
<artifactId>solon-logging-simple</artifactId>
</dependency>
<!-- metrics 指标 开始 -->
<dependency>
<groupId>org.noear</groupId>
<artifactId>solon-cloud-metrics</artifactId>
</dependency>
<dependency>
<groupId>io.micrometer</groupId>
<artifactId>micrometer-registry-prometheus</artifactId>
</dependency>
<!-- metrics 指标 结束 -->
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<scope>provided</scope>
</dependency>
</dependencies>
</project>

19
example/mica-mqtt-server-solon-plugin-example/src/main/java/org/dromara/mica/mqtt/server/solon/MqttServerApplication.java
View File

@@ -1,19 +0,0 @@
package org.dromara.mica.mqtt.server.solon;
import org.noear.solon.Solon;
import org.noear.solon.annotation.Configuration;
import org.noear.solon.scheduling.annotation.EnableScheduling;
/**
* @author wsq
*/
@Configuration
@EnableScheduling
public class MqttServerApplication {
public static void main(String[] args) {
Solon.start(MqttServerApplication.class, args);
}
}

22
example/mica-mqtt-server-solon-plugin-example/src/main/java/org/dromara/mica/mqtt/server/solon/controller/ServerController.java
View File

@@ -1,22 +0,0 @@
package org.dromara.mica.mqtt.server.solon.controller;
import org.dromara.mica.mqtt.server.solon.service.ServerService;
import org.noear.solon.annotation.Controller;
import org.noear.solon.annotation.Inject;
import org.noear.solon.annotation.Mapping;
import org.noear.solon.annotation.Post;
@Mapping("/mqtt/server")
@Controller
public class ServerController {
@Inject
private ServerService service;
@Mapping("publish")
@Post
public boolean publish(String body) {
return service.publish(body);
}
}

37
example/mica-mqtt-server-solon-plugin-example/src/main/java/org/dromara/mica/mqtt/server/solon/listener/MqttConnectOfflineListener.java
View File

@@ -1,37 +0,0 @@
/*
* 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 org.dromara.mica.mqtt.server.solon.listener;
import lombok.extern.slf4j.Slf4j;
import org.dromara.mica.mqtt.server.solon.event.MqttClientOfflineEvent;
import org.noear.solon.annotation.Component;
import org.noear.solon.core.event.EventListener;
/**
* mqtt 连接状态，使用 solon event 方式，性能有损耗
*
* @author L.cm
*/
@Slf4j
@Component
public class MqttConnectOfflineListener implements EventListener<MqttClientOfflineEvent> {
@Override
public void onEvent(MqttClientOfflineEvent mqttClientOfflineEvent) throws Throwable {
log.info("MqttClientOnlineEvent:{}", mqttClientOfflineEvent);
}
}

37
example/mica-mqtt-server-solon-plugin-example/src/main/java/org/dromara/mica/mqtt/server/solon/listener/MqttConnectOnlineListener.java
View File

@@ -1,37 +0,0 @@
/*
* 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 org.dromara.mica.mqtt.server.solon.listener;
import lombok.extern.slf4j.Slf4j;
import org.dromara.mica.mqtt.server.solon.event.MqttClientOnlineEvent;
import org.noear.solon.annotation.Component;
import org.noear.solon.core.event.EventListener;
/**
* mqtt 连接状态，使用 solon event 方式，性能有损耗
*
* @author L.cm
*/
@Slf4j
@Component
public class MqttConnectOnlineListener implements EventListener<MqttClientOnlineEvent> {
@Override
public void onEvent(MqttClientOnlineEvent mqttClientOnlineEvent) throws Throwable {
log.info("MqttClientOnlineEvent:{}", mqttClientOnlineEvent);
}
}

25
example/mica-mqtt-server-solon-plugin-example/src/main/java/org/dromara/mica/mqtt/server/solon/listener/MqttServerMessageListener1.java
View File

@@ -1,25 +0,0 @@
package org.dromara.mica.mqtt.server.solon.listener;
import lombok.extern.slf4j.Slf4j;
import org.dromara.mica.mqtt.codec.message.MqttPublishMessage;
import org.dromara.mica.mqtt.codec.MqttQoS;
import org.dromara.mica.mqtt.core.server.event.IMqttMessageListener;
import org.tio.core.ChannelContext;
import java.nio.charset.StandardCharsets;
/**
* 消息监听器示例1，直接实现 IMqttMessageListener，注意：如果实现了 IMqttMessageListener，MqttServerFunction 注解就不生效了。
*
* @author wsq
*/
@Slf4j
//@Component
public class MqttServerMessageListener1 implements IMqttMessageListener {
@Override
public void onMessage(ChannelContext context, String clientId, String topic, MqttQoS qos, MqttPublishMessage message) {
log.info("context:{} clientId:{} message:{} payload:{}", context, clientId, message, new String(message.payload(), StandardCharsets.UTF_8));
}
}

48
example/mica-mqtt-server-solon-plugin-example/src/main/java/org/dromara/mica/mqtt/server/solon/listener/MqttServerMessageListener2.java
View File

@@ -1,48 +0,0 @@
package org.dromara.mica.mqtt.server.solon.listener;
import lombok.extern.slf4j.Slf4j;
import org.dromara.mica.mqtt.codec.message.MqttPublishMessage;
import org.dromara.mica.mqtt.core.annotation.MqttServerFunction;
import org.dromara.mica.mqtt.server.solon.pojo.User;
import org.noear.solon.annotation.Component;
import org.tio.core.ChannelContext;
import org.tio.core.Node;
import java.util.Map;
/**
* 消息监听器示例2，MqttServerFunction 注解订阅，注意：如果自行实现了 IMqttMessageListener，MqttServerFunction 注解就不生效了。
*
* @author L.cm
*/
@Slf4j
@Component
public class MqttServerMessageListener2 {
@MqttServerFunction("/test/object")
public void func1(String topic, User<?> user) {
log.info("topic:{} user:{}", topic, user);
}
@MqttServerFunction("/test/client")
public void func1(String topic, byte[] message) {
log.info("topic:{} message:{}", topic, new String(message));
}
/**
* MQTT消息处理函数，匹配 mqtt Topic /test/+，如何需要匹配所以消息，请使用通配符 #
*
* @param context ChannelContext，可选参数
* @param topic 实际接收到消息的主题名称，可选参数
* @param topicVars topic 中的 ${xxxx} 变量解析（v2.5.4支持），可选参数，注意：类型必须为 Map<String, String>
* @param publishMessage 完整的MQTT发布消息对象，包含消息头和负载，可选参数
* @param message 消息负载内容，以字节数组形式提供，可选参数，也可支持对象形式，默认 json 序列化
*/
@MqttServerFunction("/test/${xxxx}")
public void func3(ChannelContext context, String topic, Map<String, String> topicVars, MqttPublishMessage publishMessage, byte[] message) {
// 获取客户端节点信息
Node clientNode = context.getClientNode();
// 记录接收到的MQTT消息信息
log.info("clientNode:{} topic:{} topicVars:{} publishMessage:{} message:{}", clientNode, topic, topicVars, publishMessage, new String(message));
}
}

19
example/mica-mqtt-server-solon-plugin-example/src/main/java/org/dromara/mica/mqtt/server/solon/pojo/User.java
View File

@@ -1,19 +0,0 @@
package org.dromara.mica.mqtt.server.solon.pojo;
import lombok.Data;
@Data
public class User<T> {
private String name;
private T girlfriend;
public static User newUser(){
User<User> user1 = new User();
user1.setName("name1");
User<User> user2 = new User();
user2.setName("name2");
user2.setGirlfriend(user1);
return user2;
}
}

24
example/mica-mqtt-server-solon-plugin-example/src/main/java/org/dromara/mica/mqtt/server/solon/service/ServerService.java
View File

@@ -1,24 +0,0 @@
package org.dromara.mica.mqtt.server.solon.service;
import lombok.extern.slf4j.Slf4j;
import org.dromara.mica.mqtt.server.solon.MqttServerTemplate;
import org.noear.solon.annotation.Component;
import org.noear.solon.annotation.Inject;
import java.nio.charset.StandardCharsets;
/**
* @author wsq
*/
@Slf4j
@Component
public class ServerService {
@Inject
private MqttServerTemplate server;
public boolean publish(String body) {
boolean result = server.publishAll("/test/123", body.getBytes(StandardCharsets.UTF_8));
log.info("Mqtt publishAll result:{}", result);
return result;
}
}

25
example/mica-mqtt-server-solon-plugin-example/src/main/java/org/dromara/mica/mqtt/server/solon/task/PublishTask.java
View File

@@ -1,25 +0,0 @@
package org.dromara.mica.mqtt.server.solon.task;
import org.dromara.mica.mqtt.server.solon.MqttServerTemplate;
import org.dromara.mica.mqtt.server.solon.pojo.User;
import org.noear.solon.annotation.Component;
import org.noear.solon.annotation.Inject;
import org.noear.solon.scheduling.annotation.Scheduled;
import java.nio.charset.StandardCharsets;
/**
* @author wsq
*/
@Component
public class PublishTask {
@Inject
private MqttServerTemplate mqttServerTemplate;
@Scheduled(fixedDelay = 1000)
public void publish() {
mqttServerTemplate.publishAll("/test/123", "mica最牛皮".getBytes(StandardCharsets.UTF_8));
mqttServerTemplate.publishAll("/test/object", User.newUser());
}
}

54
example/mica-mqtt-server-solon-plugin-example/src/main/resources/app.yml
View File

@@ -1,54 +0,0 @@
# solon 配置
solon:
logging:
appender:
console:
level: INFO
# mqtt 服务端配置
mqtt:
server:
enabled: true # 是否开启服务端，默认：true
name: Mica-Mqtt-Server # 名称，默认：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 认证
username: mica # mqtt 认证用户名
password: mica # mqtt 认证密码
debug: true # 如果开启 prometheus 指标收集建议关闭
stat-enable: true # 开启指标收集，debug 和 prometheus 开启时需要打开，默认开启，关闭节省内存
mqtt-listener: # mqtt 监听器
enable: true # 是否开启，默认：false
# ip: "0.0.0.0" # 服务端 ip 默认为空，0.0.0.0，建议不要设置
port: 1883 # 端口，默认：1883
mqtt-ssl-listener: # mqtt ssl 监听器
enable: false # 是否开启，默认：false
port: 8883 # 端口，默认：8883
ssl: # ssl 配置，必须
keystore-path: # 必须参数：ssl keystore 目录，支持 classpath:/ 路径。
keystore-pass: # 必选参数：ssl keystore 密码
truststore-path: # 可选参数：ssl 双向认证 truststore 目录，支持 classpath:/ 路径。
truststore-pass: # 可选参数：ssl 双向认证 truststore 密码
client-auth: none # 是否需要客户端认证（双向认证），默认：NONE（不需要）
ws-listener: # websocket mqtt 监听器
enable: true # 是否开启，默认：false
port: 8083 # websocket 端口，默认：8083
wss-listener: # websocket ssl mqtt 监听器
enable: false # 是否开启，默认：false
port: 8084 # 端口，默认：8084
ssl: # ssl 配置，必须
keystore-path: # 必须参数：ssl keystore 目录，支持 classpath:/ 路径。
keystore-pass: # 必选参数：ssl keystore 密码
truststore-path: # 可选参数：ssl 双向认证 truststore 目录，支持 classpath:/ 路径。
truststore-pass: # 可选参数：ssl 双向认证 truststore 密码
client-auth: none # 是否需要客户端认证（双向认证），默认：NONE（不需要）
http-listener:
enable: true
port: 18083
basic-auth: # 基础认证
enable: true
username: mica
password: mica
mcp-server: # 大模型 mcp
enable: true

2
example/mica-mqtt-server-spring-boot-example/src/main/resources/application.yml
View File

@@ -25,6 +25,8 @@ spring:
#原生配置
configuration:
map-underscore-to-camel-case: true
#打印日志
log-impl: org.apache.ibatis.logging.stdout.StdOutImpl
# mqtt 服务端配置
mqtt:

3
example/pom.xml
View File

@@ -24,11 +24,8 @@
</properties>
<modules>
<module>mica-mqtt-example</module>
<module>mica-mqtt-client-spring-boot-example</module>
<module>mica-mqtt-server-spring-boot-example</module>
<module>mica-mqtt-client-solon-plugin-example</module>
<module>mica-mqtt-server-solon-plugin-example</module>
</modules>
<dependencyManagement>