+ * Licensed under the GNU LESSER GENERAL PUBLIC LICENSE 3.0; + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + *
+ * http://www.gnu.org/licenses/lgpl.html + *
+ * 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 top.continew.starter.messaging.mqtt.annotation; + +import org.springframework.stereotype.Component; + +import java.lang.annotation.*; + +/** + * mqtt topic 监听器 + * + * @author echo + * @since 2.15.0 + */ +@Target({ElementType.TYPE}) +@Retention(RetentionPolicy.RUNTIME) +@Documented +@Inherited +@Component +public @interface MqttListener { + + /** + * 要监听的 MQTT 主题 + *
支持以下配置方式: + *
{@code
+ * 方式1: 直接指定主题
+ * @MqttListener(topic = "sensor/temperature")
+ *
+ * 方式2: 使用配置文件占位符
+ *
+ * @MqttListener(topic = "${mqtt.topic}")
+ *
+ * 方式3: 使用通配符
+ * @MqttListener(topic = "sensor/+/temperature") // 单级通配符
+ * @MqttListener(topic = "sensor/#") // 多级通配符
+ * }
+ *
+ * 通配符说明: + *
支持以下配置方式: + *
QoS 等级说明: + *
+ * Licensed under the GNU LESSER GENERAL PUBLIC LICENSE 3.0; + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + *
+ * http://www.gnu.org/licenses/lgpl.html + *
+ * 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 top.continew.starter.messaging.mqtt.autoconfigure; + +import cn.hutool.core.util.ObjectUtil; +import jakarta.annotation.PostConstruct; +import org.eclipse.paho.client.mqttv3.MqttConnectOptions; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.boot.autoconfigure.AutoConfiguration; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.context.annotation.Bean; +import org.springframework.core.env.Environment; +import org.springframework.integration.annotation.ServiceActivator; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.ExecutorChannel; +import org.springframework.integration.gateway.GatewayProxyFactoryBean; +import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory; +import org.springframework.integration.mqtt.core.MqttPahoClientFactory; +import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter; +import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler; +import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter; +import org.springframework.messaging.MessageChannel; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; +import org.springframework.util.StringUtils; +import top.continew.starter.core.constant.PropertiesConstants; +import top.continew.starter.core.constant.StringConstants; +import top.continew.starter.messaging.mqtt.autoconfigure.properties.*; +import top.continew.starter.messaging.mqtt.constant.MqttConstant; +import top.continew.starter.messaging.mqtt.handler.MqttMessageInboundHandler; +import top.continew.starter.messaging.mqtt.handler.MqttShutdownHandler; +import top.continew.starter.messaging.mqtt.msg.MqttMessageConsumer; +import top.continew.starter.messaging.mqtt.msg.MqttMessageProducer; +import top.continew.starter.messaging.mqtt.strategy.MqttOptions; +import top.continew.starter.messaging.mqtt.strategy.MqttTemplate; + +import java.util.List; +import java.util.UUID; +import java.util.concurrent.ThreadPoolExecutor; + +/** + * MQTT 自动配置类 + *
+ * 用于配置 MQTT 连接参数、入站/出站通道及消息处理器等组件。 + *
+ * + * @author echo + * @since 2.15.0 + */ +@AutoConfiguration +@EnableConfigurationProperties(MqttProperties.class) +@ConditionalOnProperty(prefix = PropertiesConstants.MESSAGING_MQTT, value = PropertiesConstants.ENABLED, havingValue = "true") +public class MqttAutoConfiguration { + + private static final Logger log = LoggerFactory.getLogger(MqttAutoConfiguration.class); + + private final MqttProperties mqttProperties; + + public MqttAutoConfiguration(MqttProperties mqttProperties) { + this.mqttProperties = mqttProperties; + } + + /** + * 配置 MQTT 客户端连接选项 + *+ * 该方法创建并配置 {@link MqttConnectOptions} 实例，用于建立 MQTT 客户端与服务器的连接。 + * 包含认证信息、连接参数、重连策略、SSL/TLS 配置以及遗嘱消息等完整配置。 + *
+ * + * @return 配置完成的 MQTT 连接选项对象 + * @see MqttConnectOptions + * @see MqttProperties + */ + @Bean + public MqttConnectOptions mqttConnectOptions() { + MqttConnectOptions mqttConnectOptions = new MqttConnectOptions(); + + // 设置 MQTT 服务器地址，支持 tcp://host:port 或 ssl://host:port 格式 + mqttConnectOptions.setServerURIs(new String[] {mqttProperties.getHost()}); + + // 设置用户名（用于认证） + mqttConnectOptions.setUserName(mqttProperties.getUsername()); + + // 设置密码（转为 char[] 以增强安全性） + mqttConnectOptions.setPassword(mqttProperties.getPassword().toCharArray()); + + // 设置连接超时时间（单位：秒），默认 30 秒 + mqttConnectOptions.setConnectionTimeout(mqttProperties.getConnectionTimeout()); + + // 设置心跳间隔（单位：秒），用于维持长连接，默认 60 秒 + mqttConnectOptions.setKeepAliveInterval(mqttProperties.getKeepAliveInterval()); + + // 设置是否清除会话 + // true：每次连接时清除上次会话状态 + // false：保留订阅信息与未确认消息 + mqttConnectOptions.setCleanSession(mqttProperties.getCleanSession()); + + // 启用自动重连机制，默认 true + mqttConnectOptions.setAutomaticReconnect(mqttProperties.getAutomaticReconnect()); + + // 设置最大重连延迟（单位：毫秒），防止频繁重连，默认 128000 毫秒 + mqttConnectOptions.setMaxReconnectDelay(mqttProperties.getMaxReconnectDelay()); + + // 设置最大允许未确认的 QoS>0 消息数量，控制并发发送能力，默认 10 + mqttConnectOptions.setMaxInflight(mqttProperties.getMaxInflight()); + + // 设置自定义 WebSocket 请求头（仅在使用 ws:// 或 wss:// 协议时生效） + mqttConnectOptions.setCustomWebSocketHeaders(mqttProperties.getCustomWebSocketHeaders()); + + // 启用 HTTPS 主机名验证（适用于 SSL/TLS） + mqttConnectOptions.setHttpsHostnameVerificationEnabled(mqttProperties.getHttpsHostnameVerificationEnabled()); + + // 设置 SSL 连接所需的客户端属性，如证书、密钥、信任库等 + mqttConnectOptions.setSSLProperties(mqttProperties.getSslClientProps()); + + // 设置关闭 ExecutorService 的超时时间（单位：秒），默认 1 秒 + mqttConnectOptions.setExecutorServiceTimeout(mqttProperties.getExecutorServiceTimeout()); + + // 设置遗嘱消息（当客户端异常断开连接时由服务器自动发送） + MqttWillProperties will = mqttProperties.getWill(); + if (ObjectUtil.isNotEmpty(will)) { + // 设置遗嘱主题 + // 设置遗嘱消息内容（字节数组） + // 设置 QoS 等级 + // 设置是否保留该消息（true 表示新订阅者会收到） + mqttConnectOptions.setWill(will.getTopic(), will.getPayload().getBytes(), will.getQos(), will + .getRetained()); + } + + return mqttConnectOptions; + } + + /** + * 配置 MQTT 客户端工厂，关联连接参数。 + */ + @Bean + public MqttPahoClientFactory mqttPahoClientFactory(MqttConnectOptions mqttConnectOptions) { + DefaultMqttPahoClientFactory clientFactory = new DefaultMqttPahoClientFactory(); + clientFactory.setConnectionOptions(mqttConnectOptions); + return clientFactory; + } + + /** + * 配置入站消息通道。 + * - 若 consumer.async = true，则使用线程池处理（异步） + * - 否则使用 DirectChannel（同步） + */ + @Bean(name = MqttConstant.MQTT_INPUT_CHANNEL_NAME) + public MessageChannel mqttInputChannel() { + MqttConsumerProperties consumer = mqttProperties.getConsumer(); + Boolean async = consumer.getAsync(); + if (Boolean.TRUE.equals(async)) { + return new ExecutorChannel(mqttConsumerExecutor()); + } else { + return new DirectChannel(); + } + } + + /** + * 配置 MQTT 消息消费处理通道（队列模式），供消费者拉取使用。 + */ + @Bean(name = MqttConstant.MQTT_OUT_BOUND_CHANNEL_NAME) + public MessageChannel consumerChannel() { + MqttProducerProperties producer = mqttProperties.getProducer(); + Boolean async = producer.getAsync(); + + if (Boolean.TRUE.equals(async)) { + return new ExecutorChannel(mqttProducerExecutor()); + } else { + return new DirectChannel(); + } + } + + /** + * 配置 MQTT 出站通道，向 broker 发送消息。 + */ + @Bean(name = MqttConstant.CONSUMER_CHANNEL_NAME) + public MessageChannel mqttOutboundChannel() { + return new DirectChannel(); + } + + /** + * 配置 MQTT 入站适配器，接收来自 MQTT broker 的消息。 + */ + @Bean + public MqttPahoMessageDrivenChannelAdapter mqttPahoMessageDrivenChannelAdapter(MqttPahoClientFactory mqttPahoClientFactory, + @Qualifier(MqttConstant.MQTT_INPUT_CHANNEL_NAME) MessageChannel mqttInputChannel, + Environment environment) { + + MqttConsumerProperties consumer = mqttProperties.getConsumer(); + String clientId = consumer.getClientId(); + if (!StringUtils.hasText(clientId)) { + clientId = getClientId(environment); + } + + MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(clientId, mqttPahoClientFactory); + adapter.setAutoStartup(consumer.getAutoStartUp()); + adapter.setOutputChannel(mqttInputChannel); + adapter.setQos(consumer.getQos()); + adapter.setConverter(new DefaultPahoMessageConverter()); + adapter.setCompletionTimeout(consumer.getCompletionTimeout()); + adapter.setDisconnectCompletionTimeout(consumer.getDisconnectCompletionTimeout()); + return adapter; + } + + /** + * 配置 MQTT 出站消息处理器，用于发布消息。 + */ + @Bean + @ServiceActivator(inputChannel = MqttConstant.MQTT_OUT_BOUND_CHANNEL_NAME) + public MqttPahoMessageHandler mqttOutbound(MqttPahoClientFactory mqttPahoClientFactory, Environment environment) { + MqttProducerProperties producer = mqttProperties.getProducer(); + String clientId = producer.getClientId(); + if (!StringUtils.hasText(clientId)) { + clientId = getClientId(environment); + } + + MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(clientId, mqttPahoClientFactory); + messageHandler.setAsync(producer.getAsync()); + messageHandler.setAsyncEvents(producer.getAsyncEvents()); + messageHandler.setDefaultTopic(producer.getDefaultTopic()); + messageHandler.setDefaultQos(producer.getDefaultQos()); + messageHandler.setConverter(new DefaultPahoMessageConverter()); + messageHandler.setDefaultRetained(producer.getDefaultRetained()); + return messageHandler; + } + + /** + * 构造封装的 MqttTemplate 工具类，提供更易用的发送/订阅能力。 + */ + @Bean + public MqttOptions mqttOptions(MqttPahoMessageDrivenChannelAdapter adapter) { + return new MqttTemplate(adapter); + } + + /** + * 构造入站消息处理器，分发到自定义监听器中。 + */ + @Bean + public MqttMessageInboundHandler mqttMessageInboundHandler(List+ * Licensed under the GNU LESSER GENERAL PUBLIC LICENSE 3.0; + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + *
+ * http://www.gnu.org/licenses/lgpl.html + *
+ * 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 top.continew.starter.messaging.mqtt.autoconfigure.properties; + +import org.springframework.boot.context.properties.NestedConfigurationProperty; +import org.springframework.integration.mqtt.core.ClientManager; +import top.continew.starter.messaging.mqtt.enums.MqttQoS; + +/** + * 消费者属性 + * + * @author echo + * @since 2.15.0 + */ +public class MqttConsumerProperties { + + /** + * MQTT 服务质量等级（QoS, Quality of Service）。 + * 0：最多一次（AT_MOST_ONCE），不保证消息到达； + * 1：至少一次（AT_LEAST_ONCE），可能重复； + * 2：只有一次（EXACTLY_ONCE），确保消息仅到达一次。 + * 默认使用 QoS 0。 + */ + private Integer qos = MqttQoS.AT_MOST_ONCE.value(); + + /** + * 消息发送完成等待超时时间（单位：毫秒）。 + * 控制发送消息时，等待 broker 响应的最大时长，超过将报错。 + * 默认值参考 ClientManager.DEFAULT_COMPLETION_TIMEOUT。 + */ + private Long completionTimeout = ClientManager.DEFAULT_COMPLETION_TIMEOUT; + + /** + * 是否自动启动客户端连接。 + * 设置为 true 则在应用启动时自动连接 MQTT 服务器并订阅 Topic。 + * 默认值为 true。 + */ + private Boolean autoStartUp = true; + + /** + * MQTT 客户端 ID。 + * 用于唯一标识客户端连接，同一 broker 下不能重复。 + * 如果为空，可能由系统自动生成。 + */ + private String clientId; + + /** + * 是否启用异步消息发送。 + * 设置为 true 则消息发送不阻塞当前线程，适用于高吞吐场景； + * 设置为 false 则同步发送，便于确认是否成功。 + * 默认值为 false。 + */ + private Boolean async = false; + + /** + * 客户端断开连接时的完成超时时间（单位：毫秒）。 + * 用于控制断连操作的最长等待时间。 + * 默认值参考 ClientManager.DISCONNECT_COMPLETION_TIMEOUT。 + */ + private Long disconnectCompletionTimeout = ClientManager.DISCONNECT_COMPLETION_TIMEOUT; + + /** + * MQTT 消息处理线程池配置。 + * 包含核心线程数、最大线程数、队列容量等，控制消费者消息处理能力。 + * 默认配置为 new MqttExecutorProperties()。 + */ + @NestedConfigurationProperty + private MqttExecutorProperties executor = new MqttExecutorProperties(); + + public Integer getQos() { + return qos; + } + + public void setQos(Integer qos) { + this.qos = qos; + } + + public Long getCompletionTimeout() { + return completionTimeout; + } + + public void setCompletionTimeout(Long completionTimeout) { + this.completionTimeout = completionTimeout; + } + + public Boolean getAutoStartUp() { + return autoStartUp; + } + + public void setAutoStartUp(Boolean autoStartUp) { + this.autoStartUp = autoStartUp; + } + + public String getClientId() { + return clientId; + } + + public void setClientId(String clientId) { + this.clientId = clientId; + } + + public Boolean getAsync() { + return async; + } + + public void setAsync(Boolean async) { + this.async = async; + } + + public Long getDisconnectCompletionTimeout() { + return disconnectCompletionTimeout; + } + + public void setDisconnectCompletionTimeout(Long disconnectCompletionTimeout) { + this.disconnectCompletionTimeout = disconnectCompletionTimeout; + } + + public MqttExecutorProperties getExecutor() { + return executor; + } + + public void setExecutor(MqttExecutorProperties executor) { + this.executor = executor; + } +} \ No newline at end of file diff --git a/continew-starter-messaging/continew-starter-messaging-mqtt/src/main/java/top/continew/starter/messaging/mqtt/autoconfigure/properties/MqttExecutorProperties.java b/continew-starter-messaging/continew-starter-messaging-mqtt/src/main/java/top/continew/starter/messaging/mqtt/autoconfigure/properties/MqttExecutorProperties.java new file mode 100644 index 00000000..bc3292bc --- /dev/null +++ b/continew-starter-messaging/continew-starter-messaging-mqtt/src/main/java/top/continew/starter/messaging/mqtt/autoconfigure/properties/MqttExecutorProperties.java @@ -0,0 +1,85 @@ +/* + * Copyright (c) 2022-present Charles7c Authors. All Rights Reserved. + *
+ * Licensed under the GNU LESSER GENERAL PUBLIC LICENSE 3.0; + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + *
+ * http://www.gnu.org/licenses/lgpl.html + *
+ * 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 top.continew.starter.messaging.mqtt.autoconfigure.properties; + +/** + * 连接池属性 + * + * @author echo + * @since 2.15.0 + */ +public class MqttExecutorProperties { + + /** + * 线程池核心线程数，表示即使线程处于空闲状态，也会保留的最小线程数。 + * 通常设置为常规负载下的并发处理数量，默认值为 5。 + */ + private Integer corePoolSize = 5; + + /** + * 线程池最大线程数，表示线程池允许创建的最大线程数量。 + * 超过 corePoolSize 后，如果任务队列已满，会继续创建线程直到该值。 + * 默认值为 10。 + */ + private Integer maxPoolSize = 10; + + /** + * 线程池中线程最大空闲时间（单位：秒）。 + * 当线程数量超过 corePoolSize 且处于空闲状态超过该时间时会被销毁。 + * 默认值为 60 秒。 + */ + private Integer keepAliveSeconds = 60; + + /** + * 线程池的任务队列容量。 + * 用于缓冲提交但尚未执行的任务，超过该容量时新任务会触发拒绝策略。 + * 默认值为 512。 + */ + private Integer queueCapacity = 512; + + public Integer getCorePoolSize() { + return corePoolSize; + } + + public void setCorePoolSize(Integer corePoolSize) { + this.corePoolSize = corePoolSize; + } + + public Integer getMaxPoolSize() { + return maxPoolSize; + } + + public void setMaxPoolSize(Integer maxPoolSize) { + this.maxPoolSize = maxPoolSize; + } + + public Integer getKeepAliveSeconds() { + return keepAliveSeconds; + } + + public void setKeepAliveSeconds(Integer keepAliveSeconds) { + this.keepAliveSeconds = keepAliveSeconds; + } + + public Integer getQueueCapacity() { + return queueCapacity; + } + + public void setQueueCapacity(Integer queueCapacity) { + this.queueCapacity = queueCapacity; + } +} \ No newline at end of file diff --git a/continew-starter-messaging/continew-starter-messaging-mqtt/src/main/java/top/continew/starter/messaging/mqtt/autoconfigure/properties/MqttProducerProperties.java b/continew-starter-messaging/continew-starter-messaging-mqtt/src/main/java/top/continew/starter/messaging/mqtt/autoconfigure/properties/MqttProducerProperties.java new file mode 100644 index 00000000..4af4b4c4 --- /dev/null +++ b/continew-starter-messaging/continew-starter-messaging-mqtt/src/main/java/top/continew/starter/messaging/mqtt/autoconfigure/properties/MqttProducerProperties.java @@ -0,0 +1,141 @@ +/* + * Copyright (c) 2022-present Charles7c Authors. All Rights Reserved. + *
+ * Licensed under the GNU LESSER GENERAL PUBLIC LICENSE 3.0; + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + *
+ * http://www.gnu.org/licenses/lgpl.html + *
+ * 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 top.continew.starter.messaging.mqtt.autoconfigure.properties; + +import org.springframework.boot.context.properties.NestedConfigurationProperty; +import top.continew.starter.messaging.mqtt.enums.MqttQoS; + +/** + * 生产者属性 + * + * @author echo + * @since 2.15.0 + */ +public class MqttProducerProperties { + + /** + * 默认的消息服务质量等级（QoS, Quality of Service）。 + * 0：最多一次（AT_MOST_ONCE），不保证送达； + * 1：至少一次（AT_LEAST_ONCE），可能重复； + * 2：只有一次（EXACTLY_ONCE），确保仅送达一次。 + * 可在发送时指定不同 QoS，此为默认值。 + * 默认值：0（AT_MOST_ONCE）。 + */ + private Integer defaultQos = MqttQoS.AT_MOST_ONCE.value(); + + /** + * MQTT 客户端 ID。 + * 用于标识连接的唯一客户端，在 broker 中必须唯一； + * 若为空，通常系统会自动生成。 + */ + private String clientId; + + /** + * 默认发布的 Topic。 + * 当未指定 topic 时使用此 topic 发送消息。 + * 默认值："producer"。 + */ + private String defaultTopic = "producer"; + + /** + * 是否启用异步发送模式。 + * true 表示消息发送不会阻塞当前线程； + * false 表示同步等待发送完成。 + * 默认值为 false。 + */ + private Boolean async = false; + + /** + * 是否异步触发发送事件（例如发送回调等）。 + * 仅在 `async = true` 时生效； + * 设置为 true 可提升事件处理性能。 + * 默认值为 false。 + */ + private Boolean asyncEvents = false; + + /** + * 是否设置消息为保留（Retained）消息。 + * Retained 消息在发送后 broker 会保留并在新订阅者订阅时立刻推送； + * 可用于设备初始状态等场景。 + * 默认值为 false（不保留）。 + */ + private Boolean defaultRetained = false; + + /** + * MQTT 消息处理线程池配置。 + * 包含核心线程数、最大线程数、队列容量等，控制消费者消息处理能力。 + * 默认配置为 new MqttExecutorProperties()。 + */ + @NestedConfigurationProperty + private MqttExecutorProperties executor = new MqttExecutorProperties(); + + public Integer getDefaultQos() { + return defaultQos; + } + + public void setDefaultQos(Integer defaultQos) { + this.defaultQos = defaultQos; + } + + public String getClientId() { + return clientId; + } + + public void setClientId(String clientId) { + this.clientId = clientId; + } + + public String getDefaultTopic() { + return defaultTopic; + } + + public void setDefaultTopic(String defaultTopic) { + this.defaultTopic = defaultTopic; + } + + public Boolean getAsync() { + return async; + } + + public void setAsync(Boolean async) { + this.async = async; + } + + public Boolean getAsyncEvents() { + return asyncEvents; + } + + public void setAsyncEvents(Boolean asyncEvents) { + this.asyncEvents = asyncEvents; + } + + public Boolean getDefaultRetained() { + return defaultRetained; + } + + public void setDefaultRetained(Boolean defaultRetained) { + this.defaultRetained = defaultRetained; + } + + public MqttExecutorProperties getExecutor() { + return executor; + } + + public void setExecutor(MqttExecutorProperties executor) { + this.executor = executor; + } +} diff --git a/continew-starter-messaging/continew-starter-messaging-mqtt/src/main/java/top/continew/starter/messaging/mqtt/autoconfigure/properties/MqttProperties.java b/continew-starter-messaging/continew-starter-messaging-mqtt/src/main/java/top/continew/starter/messaging/mqtt/autoconfigure/properties/MqttProperties.java new file mode 100644 index 00000000..51f5b769 --- /dev/null +++ b/continew-starter-messaging/continew-starter-messaging-mqtt/src/main/java/top/continew/starter/messaging/mqtt/autoconfigure/properties/MqttProperties.java @@ -0,0 +1,288 @@ +/* + * Copyright (c) 2022-present Charles7c Authors. All Rights Reserved. + *
+ * Licensed under the GNU LESSER GENERAL PUBLIC LICENSE 3.0; + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + *
+ * http://www.gnu.org/licenses/lgpl.html + *
+ * 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 top.continew.starter.messaging.mqtt.autoconfigure.properties; + +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.boot.context.properties.NestedConfigurationProperty; +import top.continew.starter.core.constant.PropertiesConstants; + +import javax.net.ssl.HostnameVerifier; +import java.util.Properties; + +/** + * 配置参数 + * + * @author echo + * @since 2.15.0 + */ +@ConfigurationProperties(prefix = PropertiesConstants.MESSAGING_MQTT) +public class MqttProperties { + + /** + * 开关 + */ + private boolean enabled; + + /** + * 地址 格式 tcp://192.168.20.95:1883 + */ + private String host; + + /** + * 用户名 + */ + private String username; + + /** + * 密码 + */ + private String password; + + /** + * 保持连接的间隔时间（秒）。客户端会按照此间隔向服务器发送心跳，以维持连接。 + * 默认值：60 秒。 + */ + private Integer keepAliveInterval = 60; + + /** + * 客户端允许同时存在的最大未确认消息数。 + * 如果超出此数量，新的消息将被阻塞直到有确认消息返回。 + * 默认值：10。 + */ + private Integer maxInflight = 10; + + /** + * 遗嘱消息内容。客户端异常断开连接时，由服务器向指定主题发送的消息。 + * 可设置 topic payload、QOS、retained 等属性。 + */ + @NestedConfigurationProperty + private MqttWillProperties will; + + /** + * 配置 SSL 连接所需的客户端属性。 + * 例如：证书路径、密钥密码等。 + */ + @NestedConfigurationProperty + private Properties sslClientProps; + + /** + * 是否启用 HTTPS 主机名验证。 + * 若为 true，将校验服务端证书中的主机名是否与实际连接地址一致。 + * 默认值：false。 + */ + private Boolean httpsHostnameVerificationEnabled = false; + + /** + * 自定义的主机名校验器，用于验证 SSL/TLS 连接时服务端主机名是否合法。 + * 可用于替换默认的验证策略。 + */ + private HostnameVerifier sslHostnameVerifier; + + /** + * 是否使用清洁会话。 + * true 表示连接建立时清除之前的会话信息（订阅、未送达消息等）， + * false 表示会话持久化。 + * 默认值：false（持久会话）。 + */ + private Boolean cleanSession = false; + + /** + * 连接超时时间（秒）。客户端尝试连接服务器的最长等待时间。 + * 默认值：30 秒。 + */ + private Integer connectionTimeout = 30; + + /** + * 是否启用自动重连。当连接断开时，是否自动尝试重新连接。 + * 默认值：true。 + */ + private Boolean automaticReconnect = true; + + /** + * 最大重连延迟时间（毫秒）。用于自动重连时的退避策略上限。 + * 默认值：128000（约 2 分钟）。 + */ + private Integer maxReconnectDelay = 128000; + + /** + * 自定义 WebSocket 请求头。 + * 用于配置使用 WebSocket 协议连接时的额外 HTTP 请求头参数。 + */ + @NestedConfigurationProperty + private Properties customWebSocketHeaders; + + /** + * 终止执行服务时等待多长时间（以秒为单位） + */ + private Integer executorServiceTimeout = 1; + + /** + * 生产者 + */ + @NestedConfigurationProperty + private MqttProducerProperties producer = new MqttProducerProperties(); + + /** + * 消费者 + */ + @NestedConfigurationProperty + private MqttConsumerProperties consumer = new MqttConsumerProperties(); + + public boolean isEnabled() { + return enabled; + } + + public void setEnabled(boolean enabled) { + this.enabled = enabled; + } + + public String getHost() { + return host; + } + + public void setHost(String host) { + this.host = host; + } + + public String getUsername() { + return username; + } + + public void setUsername(String username) { + this.username = username; + } + + public String getPassword() { + return password; + } + + public void setPassword(String password) { + this.password = password; + } + + public Integer getKeepAliveInterval() { + return keepAliveInterval; + } + + public void setKeepAliveInterval(Integer keepAliveInterval) { + this.keepAliveInterval = keepAliveInterval; + } + + public Integer getMaxInflight() { + return maxInflight; + } + + public void setMaxInflight(Integer maxInflight) { + this.maxInflight = maxInflight; + } + + public MqttWillProperties getWill() { + return will; + } + + public void setWill(MqttWillProperties will) { + this.will = will; + } + + public Properties getSslClientProps() { + return sslClientProps; + } + + public void setSslClientProps(Properties sslClientProps) { + this.sslClientProps = sslClientProps; + } + + public Boolean getHttpsHostnameVerificationEnabled() { + return httpsHostnameVerificationEnabled; + } + + public void setHttpsHostnameVerificationEnabled(Boolean httpsHostnameVerificationEnabled) { + this.httpsHostnameVerificationEnabled = httpsHostnameVerificationEnabled; + } + + public HostnameVerifier getSslHostnameVerifier() { + return sslHostnameVerifier; + } + + public void setSslHostnameVerifier(HostnameVerifier sslHostnameVerifier) { + this.sslHostnameVerifier = sslHostnameVerifier; + } + + public Boolean getCleanSession() { + return cleanSession; + } + + public void setCleanSession(Boolean cleanSession) { + this.cleanSession = cleanSession; + } + + public Integer getConnectionTimeout() { + return connectionTimeout; + } + + public void setConnectionTimeout(Integer connectionTimeout) { + this.connectionTimeout = connectionTimeout; + } + + public Boolean getAutomaticReconnect() { + return automaticReconnect; + } + + public void setAutomaticReconnect(Boolean automaticReconnect) { + this.automaticReconnect = automaticReconnect; + } + + public Integer getMaxReconnectDelay() { + return maxReconnectDelay; + } + + public void setMaxReconnectDelay(Integer maxReconnectDelay) { + this.maxReconnectDelay = maxReconnectDelay; + } + + public Properties getCustomWebSocketHeaders() { + return customWebSocketHeaders; + } + + public void setCustomWebSocketHeaders(Properties customWebSocketHeaders) { + this.customWebSocketHeaders = customWebSocketHeaders; + } + + public Integer getExecutorServiceTimeout() { + return executorServiceTimeout; + } + + public void setExecutorServiceTimeout(Integer executorServiceTimeout) { + this.executorServiceTimeout = executorServiceTimeout; + } + + public MqttProducerProperties getProducer() { + return producer; + } + + public void setProducer(MqttProducerProperties producer) { + this.producer = producer; + } + + public MqttConsumerProperties getConsumer() { + return consumer; + } + + public void setConsumer(MqttConsumerProperties consumer) { + this.consumer = consumer; + } +} diff --git a/continew-starter-messaging/continew-starter-messaging-mqtt/src/main/java/top/continew/starter/messaging/mqtt/autoconfigure/properties/MqttWillProperties.java b/continew-starter-messaging/continew-starter-messaging-mqtt/src/main/java/top/continew/starter/messaging/mqtt/autoconfigure/properties/MqttWillProperties.java new file mode 100644 index 00000000..9a30af79 --- /dev/null +++ b/continew-starter-messaging/continew-starter-messaging-mqtt/src/main/java/top/continew/starter/messaging/mqtt/autoconfigure/properties/MqttWillProperties.java @@ -0,0 +1,85 @@ +/* + * Copyright (c) 2022-present Charles7c Authors. All Rights Reserved. + *
+ * Licensed under the GNU LESSER GENERAL PUBLIC LICENSE 3.0; + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + *
+ * http://www.gnu.org/licenses/lgpl.html + *
+ * 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 top.continew.starter.messaging.mqtt.autoconfigure.properties; + +import top.continew.starter.messaging.mqtt.enums.MqttQoS; + +/** + * 遗嘱消息属性 + * + * @author echo + * @since 2.15.0 + */ +public class MqttWillProperties { + + /** + * 遗嘱消息的目标主题。 + * 当客户端异常断开时，将向该主题发布遗嘱消息。 + */ + private String topic; + + /** + * 遗嘱消息内容 + */ + private String payload; + + /** + * 遗嘱消息的 QoS 等级 + * 0：最多一次；1：至少一次；2：只有一次 + * 默认值：0 + */ + private Integer qos = MqttQoS.AT_MOST_ONCE.value(); + + /** + * 是否设置为保留消息 + * true：新订阅者会立即收到该消息 + * 默认值：false + */ + private Boolean retained = false; + + public String getTopic() { + return topic; + } + + public void setTopic(String topic) { + this.topic = topic; + } + + public String getPayload() { + return payload; + } + + public void setPayload(String payload) { + this.payload = payload; + } + + public Integer getQos() { + return qos; + } + + public void setQos(Integer qos) { + this.qos = qos; + } + + public Boolean getRetained() { + return retained; + } + + public void setRetained(Boolean retained) { + this.retained = retained; + } +} diff --git a/continew-starter-messaging/continew-starter-messaging-mqtt/src/main/java/top/continew/starter/messaging/mqtt/constant/MqttConstant.java b/continew-starter-messaging/continew-starter-messaging-mqtt/src/main/java/top/continew/starter/messaging/mqtt/constant/MqttConstant.java new file mode 100644 index 00000000..890666bb --- /dev/null +++ b/continew-starter-messaging/continew-starter-messaging-mqtt/src/main/java/top/continew/starter/messaging/mqtt/constant/MqttConstant.java @@ -0,0 +1,42 @@ +/* + * Copyright (c) 2022-present Charles7c Authors. All Rights Reserved. + *
+ * Licensed under the GNU LESSER GENERAL PUBLIC LICENSE 3.0; + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + *
+ * http://www.gnu.org/licenses/lgpl.html + *
+ * 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 top.continew.starter.messaging.mqtt.constant; + +/** + * mqtt常量 + * + * @author echo + * @since 2.15.0 + */ +public class MqttConstant { + + /** + * MQTT 入站通道名称（消费者使用，接收消息的入口） + */ + public static final String MQTT_INPUT_CHANNEL_NAME = "mqttInputChannel"; + + /** + * MQTT 出站通道名称（生产者使用，发送消息的出口） + */ + public static final String MQTT_OUT_BOUND_CHANNEL_NAME = "mqttOutboundChannel"; + + /** + * 应用级消息消费处理通道名称（接收到 MQTT 消息后转发到此通道供业务处理） + */ + public static final String CONSUMER_CHANNEL_NAME = "consumerChannel"; + +} diff --git a/continew-starter-messaging/continew-starter-messaging-mqtt/src/main/java/top/continew/starter/messaging/mqtt/enums/MqttQoS.java b/continew-starter-messaging/continew-starter-messaging-mqtt/src/main/java/top/continew/starter/messaging/mqtt/enums/MqttQoS.java new file mode 100644 index 00000000..4e4edbbd --- /dev/null +++ b/continew-starter-messaging/continew-starter-messaging-mqtt/src/main/java/top/continew/starter/messaging/mqtt/enums/MqttQoS.java @@ -0,0 +1,70 @@ +/* + * Copyright (c) 2022-present Charles7c Authors. All Rights Reserved. + *
+ * Licensed under the GNU LESSER GENERAL PUBLIC LICENSE 3.0; + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + *
+ * http://www.gnu.org/licenses/lgpl.html + *
+ * 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 top.continew.starter.messaging.mqtt.enums; + +import top.continew.starter.messaging.mqtt.exception.MqttException; + +/** + * qos 消息质量等级枚举 + * + * @author echo + * @since 2.15.0 + */ +public enum MqttQoS { + + /** + * QoS level 0 至多发送一次，发送即丢弃。没有确认消息，也不知道对方是否收到。 + */ + AT_MOST_ONCE(0), + /** + * QoS level 1 至少一次，都要在可变头部中附加一个16位的消息ID，SUBSCRIBE 和 UNSUBSCRIBE 消息使用 QoS level 1。 + */ + AT_LEAST_ONCE(1), + /** + * QoS level 2 确保只有一次，仅仅在 PUBLISH 类型消息中出现，要求在可变头部中要附加消息ID。 + */ + EXACTLY_ONCE(2), + /** + * 失败 + */ + FAILURE(0x80); + + private final int value; + + MqttQoS(int value) { + this.value = value; + } + + public int value() { + return value; + } + + public static MqttQoS valueOf(int value) { + return switch (value) { + case 0 -> AT_MOST_ONCE; + case 1 -> AT_LEAST_ONCE; + case 2 -> EXACTLY_ONCE; + case 0x80 -> FAILURE; + default -> throw new MqttException("无效的 QoS: " + value); + }; + } + + @Override + public String toString() { + return "QoS" + value; + } +} diff --git a/continew-starter-messaging/continew-starter-messaging-mqtt/src/main/java/top/continew/starter/messaging/mqtt/enums/TopicFilterType.java b/continew-starter-messaging/continew-starter-messaging-mqtt/src/main/java/top/continew/starter/messaging/mqtt/enums/TopicFilterType.java new file mode 100644 index 00000000..25f528d3 --- /dev/null +++ b/continew-starter-messaging/continew-starter-messaging-mqtt/src/main/java/top/continew/starter/messaging/mqtt/enums/TopicFilterType.java @@ -0,0 +1,124 @@ +/* + * Copyright (c) 2022-present Charles7c Authors. All Rights Reserved. + *
+ * Licensed under the GNU LESSER GENERAL PUBLIC LICENSE 3.0; + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + *
+ * http://www.gnu.org/licenses/lgpl.html + *
+ * 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 top.continew.starter.messaging.mqtt.enums;
+
+import top.continew.starter.messaging.mqtt.exception.MqttException;
+import top.continew.starter.messaging.mqtt.util.TopicUtils;
+
+/**
+ * topic 筛选器类型
+ *
+ * @author echo
+ * @since 2.15.0
+ */
+public enum TopicFilterType {
+
+ /**
+ * 默认 TopicFilter
+ */
+ NONE {
+ @Override
+ public boolean match(String topicFilter, String topicName) {
+ return TopicUtils.match(topicFilter, topicName);
+ }
+ },
+
+ /**
+ * $queue/ 为前缀的共享订阅是不带群组的共享订阅
+ */
+ QUEUE {
+ @Override
+ public boolean match(String topicFilter, String topicName) {
+ int prefixLen = TopicFilterType.SHARE_QUEUE_PREFIX.length();
+ return TopicUtils.match(topicFilter.substring(prefixLen), topicName);
+ }
+ },
+
+ /**
+ * $share/{group-name}/ 为前缀的共享订阅是带群组的共享订阅
+ */
+ SHARE {
+ @Override
+ public boolean match(String topicFilter, String topicName) {
+ // 去除前缀 $share/
+ * Licensed under the GNU LESSER GENERAL PUBLIC LICENSE 3.0;
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.gnu.org/licenses/lgpl.html
+ *
+ * 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 top.continew.starter.messaging.mqtt.exception;
+
+import top.continew.starter.core.exception.BaseException;
+
+import java.io.Serial;
+
+/**
+ * mqtt异常
+ *
+ * @author echo
+ * @since 2.15.0
+ */
+public class MqttException extends BaseException {
+
+ @Serial
+ private static final long serialVersionUID = 1L;
+
+ public MqttException() {
+ }
+
+ public MqttException(String message) {
+ super(message);
+ }
+
+ public MqttException(Throwable cause) {
+ super(cause);
+ }
+
+ public MqttException(String message, Throwable cause) {
+ super(message, cause);
+ }
+}
diff --git a/continew-starter-messaging/continew-starter-messaging-mqtt/src/main/java/top/continew/starter/messaging/mqtt/handler/MqttMessageInboundHandler.java b/continew-starter-messaging/continew-starter-messaging-mqtt/src/main/java/top/continew/starter/messaging/mqtt/handler/MqttMessageInboundHandler.java
new file mode 100644
index 00000000..6372d6fa
--- /dev/null
+++ b/continew-starter-messaging/continew-starter-messaging-mqtt/src/main/java/top/continew/starter/messaging/mqtt/handler/MqttMessageInboundHandler.java
@@ -0,0 +1,231 @@
+/*
+ * Copyright (c) 2022-present Charles7c Authors. All Rights Reserved.
+ *
+ * Licensed under the GNU LESSER GENERAL PUBLIC LICENSE 3.0;
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.gnu.org/licenses/lgpl.html
+ *
+ * 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 top.continew.starter.messaging.mqtt.handler;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.core.annotation.AnnotationUtils;
+import org.springframework.core.env.Environment;
+import org.springframework.integration.annotation.ServiceActivator;
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.MessagingException;
+import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+import org.springframework.util.CollectionUtils;
+import org.springframework.util.StringUtils;
+import top.continew.starter.messaging.mqtt.annotation.MqttListener;
+import top.continew.starter.messaging.mqtt.constant.MqttConstant;
+import top.continew.starter.messaging.mqtt.enums.TopicFilterType;
+import top.continew.starter.messaging.mqtt.exception.MqttException;
+import top.continew.starter.messaging.mqtt.model.MqttMessage;
+import top.continew.starter.messaging.mqtt.msg.MqttMessageConsumer;
+import top.continew.starter.messaging.mqtt.strategy.MqttOptions;
+import top.continew.starter.messaging.mqtt.util.TopicUtils;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * 消息调度器 路由分发中心
+ *
+ * @author echo
+ * @since 2.15.0
+ */
+public class MqttMessageInboundHandler implements MessageHandler, InitializingBean, ApplicationContextAware {
+
+ private static final Logger log = LoggerFactory.getLogger(MqttMessageInboundHandler.class);
+
+ // 精确匹配的topic -> 监听器映射（用于@MqttListener注解的监听器）
+ private final Map
+ * Licensed under the GNU LESSER GENERAL PUBLIC LICENSE 3.0;
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.gnu.org/licenses/lgpl.html
+ *
+ * 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 top.continew.starter.messaging.mqtt.handler;
+
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.context.ApplicationListener;
+import org.springframework.context.event.ContextClosedEvent;
+import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
+import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
+
+/**
+ * mqtt关闭处理程序
+ *
+ * @author echo
+ * @since 2.15.0
+ */
+@SuppressWarnings("ClassCanBeRecord")
+public class MqttShutdownHandler implements DisposableBean, ApplicationListener
+ * Licensed under the GNU LESSER GENERAL PUBLIC LICENSE 3.0;
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.gnu.org/licenses/lgpl.html
+ *
+ * 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 top.continew.starter.messaging.mqtt.model;
+
+import org.springframework.integration.mqtt.support.MqttHeaders;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+import top.continew.starter.messaging.mqtt.enums.MqttQoS;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+/**
+ * mqtt消息实体
+ *
+ * @author echo
+ * @since 2.15.0
+ */
+public class MqttMessage implements Serializable {
+
+ /**
+ * 消息头信息，包含 MQTT 消息的元数据，如 topic、QOS、retained 等。
+ */
+ private MessageHeaders messageHeaders;
+
+ /**
+ * 消息体（负载），可以是字符串、字节数组或其他对象类型。
+ */
+ private Object payload;
+
+ /**
+ * MQTT 主题，用于标识消息的发布/订阅通道。
+ */
+ private String topic;
+
+ /**
+ * 消息服务质量等级（QOS）：
+ * 0 - 最多一次，消息可能会丢失；
+ * 1 - 至少一次，消息可能重复；
+ * 2 - 只有一次，确保消息不重复也不丢失。
+ */
+ private Integer qos;
+
+ /**
+ * 是否保留该消息（Retained）：
+ * true 表示该消息会保留在 MQTT 服务器上，供新订阅者立即获取；
+ * false 表示仅当前订阅者接收到该消息。
+ */
+ private Boolean retained;
+
+ public MqttMessage(Message message) {
+ this.messageHeaders = message.getHeaders();
+ this.topic = (String)Objects.requireNonNull(message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC));
+ this.qos = (Integer)Objects.requireNonNull(message.getHeaders().get(MqttHeaders.RECEIVED_QOS));
+ this.retained = (Boolean)Objects.requireNonNull(message.getHeaders().get(MqttHeaders.RECEIVED_RETAINED));
+ this.payload = message.getPayload();
+ }
+
+ public MqttMessage(Object payload, String topic) {
+ this.payload = payload;
+ this.topic = topic;
+ this.qos = MqttQoS.AT_MOST_ONCE.value();
+ this.retained = false;
+ }
+
+ public MqttMessage(Object payload, String topic, Integer qos) {
+ this.payload = payload;
+ this.topic = topic;
+ this.qos = qos;
+ this.retained = false;
+ }
+
+ public MqttMessage(Object payload, String topic, Integer qos, Boolean retained) {
+ this.payload = payload;
+ this.topic = topic;
+ this.qos = qos;
+ this.retained = retained;
+ }
+
+ public static MqttMessage of(Message message) {
+ return new MqttMessage(message);
+ }
+
+ public static MqttMessage of(Object payload, String topic, Integer qos) {
+ return new MqttMessage(payload, topic, qos);
+ }
+
+ public MessageHeaders getMessageHeaders() {
+ return messageHeaders;
+ }
+
+ public void setMessageHeaders(MessageHeaders messageHeaders) {
+ this.messageHeaders = messageHeaders;
+ }
+
+ public Object getPayload() {
+ return payload;
+ }
+
+ public void setPayload(Object payload) {
+ this.payload = payload;
+ }
+
+ public String getTopic() {
+ return topic;
+ }
+
+ public void setTopic(String topic) {
+ this.topic = topic;
+ }
+
+ public Integer getQos() {
+ return qos;
+ }
+
+ public void setQos(Integer qos) {
+ this.qos = qos;
+ }
+
+ public Boolean getRetained() {
+ return retained;
+ }
+
+ public void setRetained(Boolean retained) {
+ this.retained = retained;
+ }
+}
diff --git a/continew-starter-messaging/continew-starter-messaging-mqtt/src/main/java/top/continew/starter/messaging/mqtt/msg/MqttMessageConsumer.java b/continew-starter-messaging/continew-starter-messaging-mqtt/src/main/java/top/continew/starter/messaging/mqtt/msg/MqttMessageConsumer.java
new file mode 100644
index 00000000..50bc7837
--- /dev/null
+++ b/continew-starter-messaging/continew-starter-messaging-mqtt/src/main/java/top/continew/starter/messaging/mqtt/msg/MqttMessageConsumer.java
@@ -0,0 +1,37 @@
+/*
+ * Copyright (c) 2022-present Charles7c Authors. All Rights Reserved.
+ *
+ * Licensed under the GNU LESSER GENERAL PUBLIC LICENSE 3.0;
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.gnu.org/licenses/lgpl.html
+ *
+ * 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 top.continew.starter.messaging.mqtt.msg;
+
+import top.continew.starter.messaging.mqtt.handler.MqttMessageInboundHandler;
+import top.continew.starter.messaging.mqtt.model.MqttMessage;
+
+/**
+ * 消息监听 - 消费者
+ *
+ * @author echo
+ * @since 2.15.0
+ */
+public interface MqttMessageConsumer {
+
+ /**
+ * 消息订阅
+ *
+ * @param message {@link MqttMessage}
+ * @see MqttMessageInboundHandler
+ */
+ void onMessage(MqttMessage message);
+}
diff --git a/continew-starter-messaging/continew-starter-messaging-mqtt/src/main/java/top/continew/starter/messaging/mqtt/msg/MqttMessageProducer.java b/continew-starter-messaging/continew-starter-messaging-mqtt/src/main/java/top/continew/starter/messaging/mqtt/msg/MqttMessageProducer.java
new file mode 100644
index 00000000..24e3c2c0
--- /dev/null
+++ b/continew-starter-messaging/continew-starter-messaging-mqtt/src/main/java/top/continew/starter/messaging/mqtt/msg/MqttMessageProducer.java
@@ -0,0 +1,72 @@
+/*
+ * Copyright (c) 2022-present Charles7c Authors. All Rights Reserved.
+ *
+ * Licensed under the GNU LESSER GENERAL PUBLIC LICENSE 3.0;
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.gnu.org/licenses/lgpl.html
+ *
+ * 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 top.continew.starter.messaging.mqtt.msg;
+
+import org.springframework.integration.annotation.MessagingGateway;
+import org.springframework.integration.mqtt.support.MqttHeaders;
+import org.springframework.messaging.handler.annotation.Header;
+import org.springframework.messaging.handler.annotation.Payload;
+import top.continew.starter.messaging.mqtt.constant.MqttConstant;
+
+/**
+ * 消息发送 - 生产者
+ *
+ * @author echo
+ * @since 2.15.0
+ **/
+@MessagingGateway(defaultRequestChannel = MqttConstant.MQTT_OUT_BOUND_CHANNEL_NAME)
+public interface MqttMessageProducer {
+
+ /**
+ * 消息发送 - 默认topic
+ *
+ * @param payload 消息体
+ */
+ void sendToMqtt(String payload);
+
+ /**
+ * 指定topic进行消息发送
+ *
+ * @param topic topic
+ * @param payload 消息体
+ */
+ void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Payload String payload);
+
+ /**
+ * 指定topic进行消息发送
+ *
+ * @param topic topic
+ * @param qos qos
+ * @param payload 消息体
+ */
+ void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic,
+ @Header(MqttHeaders.QOS) int qos,
+ @Header(MqttHeaders.RETAINED) boolean retained,
+ @Payload String payload);
+
+ /**
+ * 指定topic进行消息发送
+ *
+ * @param topic topic
+ * @param qos qos
+ * @param payload 消息体
+ */
+ void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic,
+ @Header(MqttHeaders.QOS) int qos,
+ @Header(MqttHeaders.RETAINED) boolean retained,
+ @Payload byte[] payload);
+}
diff --git a/continew-starter-messaging/continew-starter-messaging-mqtt/src/main/java/top/continew/starter/messaging/mqtt/strategy/MqttOptions.java b/continew-starter-messaging/continew-starter-messaging-mqtt/src/main/java/top/continew/starter/messaging/mqtt/strategy/MqttOptions.java
new file mode 100644
index 00000000..61c43371
--- /dev/null
+++ b/continew-starter-messaging/continew-starter-messaging-mqtt/src/main/java/top/continew/starter/messaging/mqtt/strategy/MqttOptions.java
@@ -0,0 +1,133 @@
+/*
+ * Copyright (c) 2022-present Charles7c Authors. All Rights Reserved.
+ *
+ * Licensed under the GNU LESSER GENERAL PUBLIC LICENSE 3.0;
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.gnu.org/licenses/lgpl.html
+ *
+ * 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 top.continew.starter.messaging.mqtt.strategy;
+
+import java.util.Collection;
+import java.util.List;
+
+/**
+ * MQTT 客户端操作接口
+ *
+ * 提供 topic 管理与消息发布能力。
+ * 实现类可基于不同 MQTT 客户端或通信方式进行扩展。
+ *
+ * Licensed under the GNU LESSER GENERAL PUBLIC LICENSE 3.0;
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.gnu.org/licenses/lgpl.html
+ *
+ * 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 top.continew.starter.messaging.mqtt.strategy;
+
+import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
+import top.continew.starter.messaging.mqtt.enums.MqttQoS;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * MQTT 客户端发布+订阅封装器
+ *
+ * @author echo
+ * @since 2.15.0
+ **/
+@SuppressWarnings("ClassCanBeRecord")
+public class MqttTemplate implements MqttOptions {
+
+ private final MqttPahoMessageDrivenChannelAdapter adapter;
+
+ public MqttTemplate(MqttPahoMessageDrivenChannelAdapter adapter) {
+ this.adapter = adapter;
+ }
+
+ @Override
+ public void addTopic(String topic) {
+ this.addTopic(topic, MqttQoS.AT_MOST_ONCE.value());
+ }
+
+ @Override
+ public void addTopic(String topic, int qos) {
+ Set
+ * Licensed under the GNU LESSER GENERAL PUBLIC LICENSE 3.0;
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.gnu.org/licenses/lgpl.html
+ *
+ * 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 top.continew.starter.messaging.mqtt.util;
+
+import top.continew.starter.messaging.mqtt.exception.MqttException;
+
+import java.util.List;
+
+/**
+ * 消息主题工具类
+ *
+ * @author echo
+ * @since 2.15.0
+ */
+public class TopicUtils {
+
+ public static final char TOPIC_WILDCARDS_ONE = '+';
+
+ public static final char TOPIC_WILDCARDS_MORE = '#';
+
+ public TopicUtils() {
+ }
+
+ /**
+ * 校验 topicFilter
+ *
+ * @param topicFilterList topicFilter 集合
+ */
+ public static void validateTopicFilter(List