跳至主要內容

mica-mqtt-server-spring-boot-starter 使用文档

如梦技术2024/12/10大约 7 分钟

科普视频

版本兼容

要求Spring boot 版本
最高4.x
最低2.1.0.RELEASE

一、添加依赖

<dependency>
    <groupId>org.dromara.mica-mqtt</groupId>
    <artifactId>mica-mqtt-server-spring-boot-starter</artifactId>
    <version>${最新版本}</version>
</dependency>

二、mqtt 服务

2.1 配置项

# mqtt 服务端配置
mqtt:
  server:
    # ------ 基础配置 ------
    enabled: true                       # 是否启用，默认：true
    name: Mica-Mqtt-Server              # 名称，默认：Mica-Mqtt-Server
    node-name:                          # 节点名称，用于处理集群
    debug: false                        # 调试日志，开启后会影响性能，开启 prometheus 时建议关闭
    stat-enable: true                   # 是否开启监控（指标收集），关闭可节省内存，默认：true
    proxy-protocol-on: false            # 开启代理协议，支持 nginx proxy_protocol on;，默认：false
    # ------ 线程池 ------
    tio-executor-size:                  # tio 编解码等线程数
    group-executor-size:                # AIO AsynchronousChannelGroup 的线程池
    mqtt-executor-size:                 # mqtt 工作线程数，默认：8或2倍CPU核心数（取较大值），消息量大时调大此参数
    shutdown-timeout-sec: 6000        # mqtt 工作线程池关闭等待超时时间，单位：秒，默认：6000（约 100 分钟，沿用 mica-net 默认值，2.6.8 开始支持）。
                                       # 该值仅控制 awaitTermination 的阻塞时长，超时不会强制中断线程；
                                       # 服务端 stop 时会按连接逐个触发 IMqttConnectStatusListener.onDisconnect，
                                       # 这些任务由 groupExecutor（默认 8~16 线程）串行处理，超时后这些任务仍会继续执行直到自然结束。
                                       # 请同步将部署环境终止宽限期（如 k8s terminationGracePeriodSeconds）调到不小于此值，否则进程会被 SIGKILL 强杀。
    # ------ mqtt 协议参数 ------
    heartbeat-timeout: 120000           # 心跳超时时间（单位：毫秒，默认：1000 * 120），设为 0 或负数表示禁用
    keepalive-backoff: 0.75             # MQTT 客户端 keepalive 系数（默认：0.75），连接超时 = keepalive * 系数 * 2，不得小于 0.5
    read-buffer-size: 8KB               # 接收数据的 buffer size，默认：8k
    max-bytes-in-message: 10MB          # 消息解析最大 bytes 长度，默认：10M
    max-client-id-length: 64            # 客户端 ID 最大长度，mqtt 3.1 规定 23，这里默认 64 以减少问题
    # ------ MQTT 5.0 服务端能力属性（CONNACK Properties）------
    properties:
      receive-maximum: 65535            # 服务端允许客户端同时处理的 QoS1/QoS2 未确认报文上限，默认：65535（2.6.8 开始支持）
      maximum-qos: 2                    # 服务端支持的最大 QoS，默认：2（2.6.8 开始支持）
      retain-available: true            # 服务端是否支持保留消息，默认：true（2.6.8 开始支持）
      maximum-packet-size: 268435456    # 服务端可处理的最大报文大小（字节），默认：268435456（2.6.8 开始支持）
      topic-alias-maximum: 0            # 服务端支持的最大主题别名数，0 表示不启用，默认：0（2.6.8 开始支持）
      wildcard-subscription-available: true # 服务端是否支持通配符订阅，默认：true（2.6.8 开始支持）
      shared-subscription-available: true   # 服务端是否支持共享订阅，默认：true（2.6.8 开始支持）
      subscription-identifier-available: false # 服务端是否支持订阅标识符，默认：false（2.6.8 开始支持）
      server-keep-alive: 0              # 服务端下发给 MQTT 5.0 客户端的 Keep Alive，0 表示不接管，默认：0（2.6.8 开始支持）
    # ------ mqtt 认证 ------
    auth:
      enable: false                     # 是否开启 mqtt 认证，默认：false
      username: mica                    # mqtt 认证账号
      password: mica                    # mqtt 认证密码
    # ------ mqtt tcp 监听器 ------
    mqtt-listener:
      enable: true                      # 是否启用，默认：false
#      ip: 0.0.0.0                      # 服务端 ip，默认为空（0.0.0.0），建议不要设置
      port: 1883                        # 端口，默认：1883
    # ------ mqtt ssl 监听器 ------
    mqtt-ssl-listener:
      enable: false                     # 是否启用，默认：false
      port: 8883                        # 端口，默认：8883
      ssl:
        keystore-path:                   # 必填：ssl keystore 证书路径，支持 classpath:/ 路径
        keystore-pass:                   # 必填：ssl keystore 密码
        truststore-path:                 # 可选：ssl 双向认证 truststore 证书路径，支持 classpath:/ 路径
        truststore-pass:                 # 可选：ssl 双向认证 truststore 密码
        client-auth: NONE                # 客户端认证类型，默认：NONE（不需要），可选 OPTIONAL / REQUIRE
    # ------ websocket mqtt 监听器 ------
    ws-listener:
      enable: true                      # 是否启用，默认：false
      port: 8083                        # websocket 端口，默认：8083
    # ------ websocket ssl mqtt 监听器 ------
    wss-listener:
      enable: false                     # 是否启用，默认：false
      port: 8084                        # 端口，默认：8084
      ssl:
        keystore-path:                   # 必填：ssl keystore 证书路径，支持 classpath:/ 路径
        keystore-pass:                   # 必填：ssl keystore 密码
        truststore-path:                 # 可选：ssl 双向认证 truststore 证书路径，支持 classpath:/ 路径
        truststore-pass:                 # 可选：ssl 双向认证 truststore 密码
        client-auth: NONE                # 客户端认证类型，默认：NONE（不需要），可选 OPTIONAL / REQUIRE
    # ------ http api 监听器 ------
    http-listener:
      enable: true                      # 是否启用，默认：false
      port: 18083                       # http 端口
#      ip: 0.0.0.0                      # 服务端 ip，默认为空（0.0.0.0）
      basic-auth:                       # http Basic 认证
        enable: false                   # 是否启用，默认：false
        username: mica                  # http Basic 认证账号
        password: mica                  # http Basic 认证密码
      mcp:                              # 大模型 MCP（Model Context Protocol）
        enable: false                   # 是否启用，默认：false
        endpoint: /mcp/message          # stream http endpoint
        sse-endpoint: /mcp/sse          # sse 端点
        sse-message-endpoint: /mcp/message  # sse message 端点
      ssl:                              # http ssl 配置
        enable: false                   # 是否启用，默认：false
        keystore-path:                  # 必填：ssl keystore 证书路径，支持 classpath:/ 路径
        keystore-pass:                  # 必填：ssl keystore 密码
        truststore-path:                # 可选：ssl 双向认证 truststore 证书路径
        truststore-pass:                # 可选：ssl 双向认证 truststore 密码
        client-auth: NONE               # 客户端认证类型，默认：NONE（不需要），可选 OPTIONAL / REQUIRE

注意：ssl 存在三种情况

服务端开启ssl客户端
ClientAuth 为 NONE（不需要客户端验证）仅仅需要开启 ssl 即可不用配置证书
ClientAuth 为 OPTIONAL（与客户端协商）需开启 ssl 并且配置 truststore 证书
ClientAuth 为 REQUIRE (必须的客户端验证)需开启 ssl 并且配置 truststore、 keystore证书

2.2 可实现接口（注册成 Spring Bean 即可）

接口是否必须说明
IMqttServerUniqueIdService否用于 clientId 不唯一时，自定义实现唯一标识，后续接口使用它替代 clientId
IMqttServerAuthHandler是用于服务端认证
IMqttServerSubscribeValidator否（建议实现）1.1.3 新增，用于对客户端订阅校验
IMqttServerPublishPermission否（建议实现）1.2.2 新增，用于对客户端发布权限校验
IMqttMessageListener否（1.3.x为否）消息监听
IMqttConnectStatusListener是连接状态监听
IMqttSessionManager否session 管理
IMqttSessionListener否session 监听
IMqttMessageStore集群是，单机否遗嘱和保留消息存储
AbstractMqttMessageDispatcher集群是，单机否消息转发，（遗嘱、保留消息转发）
IMqttMessageInterceptor否消息拦截器，1.3.9 新增

2.3 MqttServerFunction 注解监听客户端上传的消息使用示例（v2.5.3开始支持）

/**
 *  注解消息监听，注意：如果自行实现了 IMqttMessageListener，MqttServerFunction 注解就不生效了。
 */
@Slf4j
@Service
public class MqttServerMessageListener {

	/**
	 * MQTT消息处理函数
	 *
	 * @param topic mqtt Topic
	 * @param user  订阅消息的负载内容，默认 json 序列化
	 */
	@MqttServerFunction("/test/object")
	public void func1(String topic, User<?> user) {
		log.info("topic:{} user:{}", topic, user);
	}

	@MqttServerFunction("/test/client")
	public void func2(String topic, byte[] message) {
		log.info("topic:{} message:{}", topic, new String(message));
	}

	/**
	 * MQTT消息处理函数，匹配 mqtt Topic /test/+，如何需要匹配所以消息，请使用通配符 #
	 *
	 * @param context        ChannelContext，可选参数
	 * @param topic          实际接收到消息的主题名称，可选参数
	 * @param publishMessage 完整的MQTT发布消息对象，包含消息头和负载，可选参数
	 * @param message        消息负载内容，以字节数组形式提供，可选参数，也可支持对象形式，默认 json 序列化
	 */
	@MqttServerFunction("/test/${xxxx}")
	public void func3(ChannelContext context, String topic, MqttPublishMessage publishMessage, byte[] message) {
		// 获取客户端节点信息
		Node clientNode = context.getClientNode();
		// 记录接收到的MQTT消息信息
		log.info("clientNode:{} topic:{} publishMessage:{} message:{}", clientNode, topic, publishMessage, new String(message));
	}

}

2.3 传统 IMqttMessageListener (用于监听客户端上传的消息) 使用示例

@Service
public class MqttServerMessageListener implements IMqttMessageListener {
    private static final Logger logger = LoggerFactory.getLogger(MqttServerMessageListener.class);

    @Override
    public void onMessage(ChannelContext context, String clientId, String topic, MqttQoS qoS, MqttPublishMessage message) {
        log.info("clientId:{} message:{} payload:{}", clientId, message, new String(message.getPayload(), StandardCharsets.UTF_8));
    }
}

2.4 自定义配置（可选）

@Configuration(proxyBeanMethods = false)
public class MqttServerCustomizerConfiguration {

	@Bean
	public MqttServerCustomizer mqttServerCustomizer() {
		return new MqttServerCustomizer() {
			@Override
			public void customize(MqttServerCreator creator) {
				// 此处可自定义配置 creator，会覆盖 yml 中的配置
				System.out.println("----------------MqttServerCustomizer-----------------");
			}
		};
	}

}

2.5 MqttServerTemplate 使用示例


import org.dromara.mica.mqtt.spring.server.MqttServerTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

/**
 * @author wsq
 */
@Service
public class ServerService {
    @Autowired
    private MqttServerTemplate server;

    public boolean publish(String body) {
        server.publishAll("/test/123", body.getBytes(StandardCharsets.UTF_8));
        return true;
    }
}

2.6 客户端上下线监听

使用 Spring event 解耦客户端上下线监听，注意： 1.3.4 开始支持。会跟自定义的 IMqttConnectStatusListener 实现冲突，取一即可。

@Service
public class MqttConnectStatusListener {
	private static final Logger logger = LoggerFactory.getLogger(MqttConnectStatusListener.class);

	@EventListener
	public void online(MqttClientOnlineEvent event) {
		logger.info("MqttClientOnlineEvent:{}", event);
	}

	@EventListener
	public void offline(MqttClientOfflineEvent event) {
		logger.info("MqttClientOfflineEvent:{}", event);
	}

}

2.7 基于 mq 消息广播集群处理

详见: mica-mqtt-broker

2.8 Prometheus + Grafana 监控对接

<!-- 开启 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>
支持得指标说明
mqtt_connections_accepted共接受过连接数
mqtt_connections_closed关闭过的连接数
mqtt_connections_size当前连接数
mqtt_messages_handled_packets已处理消息数
mqtt_messages_handled_bytes已处理消息字节数
mqtt_messages_received_packets已接收消息数
mqtt_messages_received_bytes已处理消息字节数
mqtt_messages_send_packets已发送消息数
mqtt_messages_send_bytes已发送消息字节数
编辑此页
最近更新:
贡献者: 如梦技术
Copyright © 2026 如梦技术