✨ 添加 mica-mqtt-broker，待完善。

mica-mqtt-broker/pom.xml

+<?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-broker</artifactId>
+    <name>${artifactId}</name>
+    <url>https://www.dreamlu.net</url>
+
+    <parent>
+        <groupId>net.dreamlu</groupId>
+        <artifactId>mica-mqtt</artifactId>
+        <version>${revision}</version>
+    </parent>
+
+    <properties>
+        <java.version>1.8</java.version>
+        <mica.version>2.5.4</mica.version>
+        <spring.boot.version>2.5.4</spring.boot.version>
+    </properties>
+
+    <dependencies>
+        <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-redis</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>net.dreamlu</groupId>
+            <artifactId>mica-logging</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>net.dreamlu</groupId>
+            <artifactId>mica-swagger</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>net.dreamlu</groupId>
+            <artifactId>mica-mqtt-spring-boot-starter</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>
+    </dependencies>
+
+    <dependencyManagement>
+        <dependencies>
+            <dependency>
+                <groupId>net.dreamlu</groupId>
+                <artifactId>mica-bom</artifactId>
+                <version>${mica.version}</version>
+                <type>pom</type>
+                <scope>import</scope>
+            </dependency>
+            <dependency>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-dependencies</artifactId>
+                <version>${spring.boot.version}</version>
+                <type>pom</type>
+                <scope>import</scope>
+            </dependency>
+        </dependencies>
+    </dependencyManagement>
+
+    <build>
+        <finalName>${project.name}</finalName>
+        <plugins>
+            <plugin>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-maven-plugin</artifactId>
+            </plugin>
+        </plugins>
+    </build>
+</project>

mica-mqtt-broker/src/main/java/net/dreamlu/iot/mqtt/broker/MqttBrokerApplication.java

+/*
+ * 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 net.dreamlu.iot.mqtt.broker;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.cache.annotation.EnableCaching;
+import org.springframework.scheduling.annotation.EnableScheduling;
+
+/**
+ * mica mqtt broker
+ *
+ * @author L.cm
+ */
+@SpringBootApplication
+@EnableScheduling
+@EnableCaching
+public class MqttBrokerApplication {
+
+	public static void main(String[] args) {
+		SpringApplication.run(MqttBrokerApplication.class, args);
+	}
+
+}

mica-mqtt-broker/src/main/java/net/dreamlu/iot/mqtt/broker/auth/MqttAuthHandler.java

+/*
+ * 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 net.dreamlu.iot.mqtt.broker.auth;
+
+import net.dreamlu.iot.mqtt.codec.MqttQoS;
+import net.dreamlu.iot.mqtt.codec.MqttTopicSubscription;
+import net.dreamlu.iot.mqtt.core.server.IMqttServerAuthHandler;
+import org.springframework.context.annotation.Configuration;
+
+import java.util.List;
+
+/**
+ * mqtt tcp websocket 认证
+ *
+ * @author L.cm
+ */
+@Configuration(proxyBeanMethods = false)
+public class MqttAuthHandler implements IMqttServerAuthHandler {
+
+	@Override
+	public boolean authenticate(String clientId, String userName, String password) {
+		// 客户端认证逻辑实现
+		return true;
+	}
+
+	@Override
+	public boolean isValidSubscribe(List<MqttTopicSubscription> topicSubscriptionList) {
+		// 校验客户端订阅的 topic，校验成功返回 true，失败返回 false
+		for (MqttTopicSubscription subscription : topicSubscriptionList) {
+			String topicName = subscription.topicName();
+			MqttQoS mqttQoS = subscription.qualityOfService();
+		}
+		return true;
+	}
+}

mica-mqtt-broker/src/main/java/net/dreamlu/iot/mqtt/broker/auth/MqttHttpAuthFilter.java

+/*
+ * 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 net.dreamlu.iot.mqtt.broker.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 L.cm
+ */
+@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, HttpResponse response) {
+		// 认证不通过时的响应
+		return Result.fail(response, ResultCode.E103);
+	}
+
+	@Override
+	public void afterPropertiesSet() throws Exception {
+		MqttHttpRoutes.addFilter(this);
+	}
+}

mica-mqtt-broker/src/main/java/net/dreamlu/iot/mqtt/broker/config/MqttBrokerConfiguration.java

+/*
+ * 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 net.dreamlu.iot.mqtt.broker.config;
+
+import net.dreamlu.iot.mqtt.broker.listener.MqttBrokerMessageListener;
+import net.dreamlu.iot.mqtt.core.server.dispatcher.IMqttMessageDispatcher;
+import net.dreamlu.iot.mqtt.core.server.support.DefaultMqttMessageDispatcher;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * mica mqtt broker 配置
+ *
+ * @author L.cm
+ */
+@Configuration(proxyBeanMethods = false)
+public class MqttBrokerConfiguration {
+
+	@Bean
+	public IMqttMessageDispatcher messageDispatcher() {
+		// TODO L.cm 此处采用 redis 实现广播
+		return new DefaultMqttMessageDispatcher();
+	}
+
+	@Bean
+	public MqttBrokerMessageListener brokerMessageListener(IMqttMessageDispatcher dispatcher) {
+		return new MqttBrokerMessageListener(dispatcher);
+	}
+
+}

mica-mqtt-broker/src/main/java/net/dreamlu/iot/mqtt/broker/config/package-info.java

+package net.dreamlu.iot.mqtt.broker.config;

mica-mqtt-broker/src/main/java/net/dreamlu/iot/mqtt/broker/listener/MqttBrokerConnectListener.java

+/*
+ * 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 net.dreamlu.iot.mqtt.broker.listener;
+
+import net.dreamlu.iot.mqtt.core.server.event.IMqttConnectStatusListener;
+
+/**
+ * mqtt 连接监听
+ *
+ * @author L.cm
+ */
+public class MqttBrokerConnectListener implements IMqttConnectStatusListener {
+	@Override
+	public void online(String clientId) {
+
+	}
+
+	@Override
+	public void offline(String clientId) {
+
+	}
+}

mica-mqtt-broker/src/main/java/net/dreamlu/iot/mqtt/broker/listener/MqttBrokerMessageListener.java

+/*
+ * 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 net.dreamlu.iot.mqtt.broker.listener;
+
+import net.dreamlu.iot.mqtt.codec.MqttMessageType;
+import net.dreamlu.iot.mqtt.codec.MqttQoS;
+import net.dreamlu.iot.mqtt.core.server.dispatcher.IMqttMessageDispatcher;
+import net.dreamlu.iot.mqtt.core.server.event.IMqttMessageListener;
+import net.dreamlu.iot.mqtt.core.server.model.Message;
+
+import java.nio.ByteBuffer;
+import java.util.Objects;
+
+/**
+ * 集群消息监听器
+ *
+ * @author L.cm
+ */
+public class MqttBrokerMessageListener implements IMqttMessageListener {
+	private final IMqttMessageDispatcher dispatcher;
+
+	public MqttBrokerMessageListener(IMqttMessageDispatcher dispatcher) {
+		this.dispatcher = Objects.requireNonNull(dispatcher, "MqttMessageDispatcher is null.");
+	}
+
+	@Override
+	public void onMessage(String clientId, String topic, MqttQoS mqttQoS, ByteBuffer payload) {
+		Message message = new Message();
+		message.setTopic(topic);
+		message.setQos(mqttQoS.value());
+		if (payload != null) {
+			message.setPayload(payload.array());
+		}
+		message.setMessageType(MqttMessageType.PUBLISH.value());
+		message.setStoreTime(System.currentTimeMillis());
+		dispatcher.send(message);
+	}
+}

mica-mqtt-broker/src/main/resources/application-dev.yml

+mqtt:
+  server:
+    enabled: true               # 是否开启服务端，默认：true
+#    ip: 0.0.0.0                 # 服务端 ip 默认为空，0.0.0.0
+    port: 5883                  # 端口，默认：1883
+    name: Mica-Mqtt-Server      # 名称，默认：Mica-Mqtt-Server
+    buffer-allocator: HEAP      # 堆内存和堆外内存，默认：堆内存
+    heartbeat-timeout: 120000   # 心跳超时，单位毫秒，默认: 1000 * 120
+    read-buffer-size: 8092      # 接收数据的 buffer size，默认：8092
+    max-bytes-in-message: 8092  # 消息解析最大 bytes 长度，默认：8092
+    debug: true                 # 如果开启 prometheus 指标收集建议关闭
+    web-port: 8083              # http、websocket 端口，默认：8083
+    websocket-enable: true      # 是否开启 websocket，默认： true
+    http-enable: false          # 是否开启 http api，默认： false
+    http-basic-auth:
+      enable: false             # 是否开启 http basic auth，默认： false
+      username: "mica"          # http basic auth 用户名
+      password: "mica"          # http basic auth 密码

mica-mqtt-broker/src/main/resources/application.yml

+server:
+  port: 30001
+spring:
+  application:
+    name: mica-mqtt-broker
+  profiles:
+    active: dev
+# 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 服务端默认日志

mica-mqtt-broker/src/main/resources/banner.txt

+
+${AnsiColor.BRIGHT_BLUE}##     ## ####  ######     ###   ${AnsiColor.RED}    ##     ##  #######  ######## ########
+${AnsiColor.BRIGHT_BLUE}###   ###  ##  ##    ##   ## ##  ${AnsiColor.RED}    ###   ### ##     ##    ##       ##
+${AnsiColor.BRIGHT_BLUE}#### ####  ##  ##        ##   ## ${AnsiColor.RED}    #### #### ##     ##    ##       ##
+${AnsiColor.BRIGHT_BLUE}## ### ##  ##  ##       ##     ##${AnsiColor.RED}    ## ### ## ##     ##    ##       ##
+${AnsiColor.BRIGHT_BLUE}##     ##  ##  ##       #########${AnsiColor.RED}    ##     ## ##  ## ##    ##       ##
+${AnsiColor.BRIGHT_BLUE}##     ##  ##  ##    ## ##     ##${AnsiColor.RED}    ##     ## ##    ##     ##       ##
+${AnsiColor.BRIGHT_BLUE}##     ## ####  ######  ##     ##${AnsiColor.RED}    ##     ##  ##### ##    ##       ##
+
+                      https://www.dreamlu.net
+
+${AnsiColor.BRIGHT_BLUE}:: ${spring.application.name} :: Running Spring Boot ${spring-boot.version} 🏃🏃🏃 ${AnsiColor.DEFAULT}

pom.xml

@@ -28,6 +28,7 @@
         <module>mica-mqtt-spring-boot-starter</module>
         <module>mica-mqtt-example</module>
         <module>mica-mqtt-spring-boot-example</module>
+        <module>mica-mqtt-broker</module>
     </modules>
 
     <dependencyManagement>