一、引言

1.1 什么是 MQTT

MQTT 是一种基于发布/订阅范式的轻量级消息传输协议。该协议专为计算资源受限、带宽有限以及网络连接不稳定的设备设计，具有报文结构简洁、功耗低、传输可靠等特点。

1.2 为什么选择 Moquette

当业务需求比较简单没必要或开发受限无法在服务器部署第三方服务（例如 EMQX、Mosquitto、HiveMQ），那么就需要在 SpringBoot 中集成 MQTT Broker。
Moquette 是一个轻量级嵌入式 Broker，可以轻松封装在其他应用程序中。但由于0.18之前的版本不支持 MQTT5，所以这里选择 Fraunhofer 维护的新分支。

二、依赖配置

<!-- MQTT Broker：Moquette --><dependency><groupId>de.fraunhofer.iosb.io.moquette</groupId><artifactId>moquette-broker</artifactId><version>0.18.4</version></dependency>

三、代码整合

3.1 创建配置类

importio.moquette.broker.Server;importio.moquette.broker.config.IConfig;importio.moquette.broker.config.MemoryConfig;importio.moquette.interception.InterceptHandler;importjakarta.annotation.PostConstruct;importjakarta.annotation.PreDestroy;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;importjava.util.Arrays;importjava.util.List;importjava.util.Properties;@ConfigurationpublicclassMqttBrokerConfig{privateServermqttBroker;@AutowiredprivateCustomInterceptHandlerinterceptHandler;@PostConstructpublicvoidstartBroker(){mqttBroker=newServer();Propertiesprops=newProperties();props.setProperty("port","1883");//监听端口，默认 1883props.setProperty("host","0.0.0.0");//绑定地址，0.0.0.0 表示允许所有 IP 访问props.setProperty("allow_anonymous","false");//禁止匿名登录，必须用户名+密码认证（必须设为 false）//性能相关配置props.setProperty("max_inflight_queue","1000");//单个客户端最大未确认消息队列长度，防止消息堆积props.setProperty("netty.mqtt.message_size","65536");//最大消息体大小（64KB）props.setProperty("persistence_enabled","true");//是否开启持久化props.setProperty("persistence_type","file");//持久化存储类型props.setProperty("data_path","D:\\haooo\\tempFile\\mqttFile");//持久化文件存放目录props.setProperty("persistence_recover_on_startup","true");//持久化启动时自动恢复props.setProperty("max_queued_messages","10000");//最大队列消息数IConfigconfig=newMemoryConfig(props);try{List<InterceptHandler>handlers=Arrays.asList(interceptHandler);mqttBroker.startServer(config,handlers,null,deviceAuthenticator,null);System.out.println("✅ Moquette MQTT Broker 启动成功，端口 1883");}catch(Exceptione){System.err.println("❌ Moquette Broker 启动失败");e.printStackTrace();}}@PreDestroypublicvoidstopBroker(){if(mqttBroker!=null){mqttBroker.removeInterceptHandler(interceptHandler);mqttBroker.stopServer();System.out.println("✅ Moquette MQTT Broker 已停止");}}@BeanpublicServermqttServer(){returnmqttBroker;//返回已启动的实例}}

3.2 自定义认证

importcom.github.benmanes.caffeine.cache.Cache;importcom.github.benmanes.caffeine.cache.Caffeine;importio.moquette.broker.security.IAuthenticator;importorg.springframework.stereotype.Component;importjava.time.Duration;/** * 自定义认证 * Moquette 支持插件式认证，实现 IAuthenticator */@ComponentpublicclassDeviceAuthenticatorimplementsIAuthenticator{privatefinalCache<String,Boolean>authCache=Caffeine.newBuilder().expireAfterWrite(Duration.ofMinutes(5)).build();@OverridepublicbooleancheckValid(StringclientId,Stringusername,byte[]password){if(username==null||password==null||clientId==null){System.out.println("❌ 设备认证失败：用户名或密码或客户端标识为空");returnfalse;}StringpwdStr=newString(password);//验证设备是否存在（5分钟内的验证结果进行缓存，避免认证失败的设备频繁重试）Booleancached=authCache.getIfPresent(clientId+":"+username);if(cached!=null){returncached;}else{//数据库查询Devicedevice=DeviceMapper.selectDevice(username,clientId,pwdStr);if(device!=null){System.out.println("✅ 设备认证成功: "+clientId+" | "+username+" | "+pwdStr);authCache.put(clientId+":"+username,true);returntrue;}else{authCache.put(clientId+":"+username,false);}}System.out.println("❌ 设备认证失败: "+clientId+" | "+username+" | "+pwdStr);returnfalse;}}

3.3 消息处理

importio.moquette.interception.AbstractInterceptHandler;importio.moquette.interception.messages.InterceptConnectMessage;importio.moquette.interception.messages.InterceptDisconnectMessage;importio.moquette.interception.messages.InterceptPublishMessage;importio.netty.buffer.ByteBuf;importorg.springframework.stereotype.Component;importjava.nio.charset.StandardCharsets;@ComponentpublicclassCustomInterceptHandlerextendsAbstractInterceptHandler{@OverridepublicStringgetID(){return"CustomInterceptHandler";}@OverridepublicvoidonConnect(InterceptConnectMessagemsg){System.out.println("✅ 设备连接成功: "+msg.getClientID()+" | "+msg.getUsername());//TODO: 更新数据库设备在线离线状态}@OverridepublicvoidonDisconnect(InterceptDisconnectMessagemsg){System.out.println("❌ 设备连接断开: "+msg.getClientID()+" | "+msg.getUsername());//TODO: 更新数据库设备在线离线状态}@OverridepublicvoidonPublish(InterceptPublishMessagemsg){ByteBufbuf=msg.getPayload();if(buf==null||buf.readableBytes()==0){return;}byte[]bytes=newbyte[buf.readableBytes()];buf.getBytes(buf.readerIndex(),bytes);Stringpayload=newString(bytes,StandardCharsets.UTF_8);System.out.println("✅ 设备发布消息: "+msg.getClientID()+" | "+msg.getUsername()+" | "+msg.getTopicName()+" | "+payload);//处理设备发送的消息if(msg.getTopicName().matches("devices/[^/]+/[^/]+/resp")){//处理设备发布主题}else{System.out.println("❌ 未匹配设备发布消息: "+msg.getClientID()+" | "+msg.getUsername()+" | "+msg.getTopicName()+" | "+payload);}}@OverridepublicvoidonSessionLoopError(Throwablecause){System.out.println("❌ 会话循环错误: "+cause.getMessage());}}

3.4 发送指令

importcom.alibaba.fastjson2.JSONObject;importio.moquette.broker.Server;importio.netty.buffer.ByteBuf;importio.netty.buffer.Unpooled;importio.netty.handler.codec.mqtt.*;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.context.annotation.Lazy;importorg.springframework.stereotype.Service;@ServicepublicclassMqttCommandService{@Autowired@LazyprivateServermqttBroker;/** * 向指定设备发送指令 * @param clientId 客户端标识 * @param username 设备序列号 * @param command 指令内容 */publicvoidsendCommand(StringclientId,Stringusername,JSONObjectcommand){Stringtopic="devices/"+clientId+"/"+username+"/cmd";//订阅主题try{MqttFixedHeadermqttFixedHeader=newMqttFixedHeader(MqttMessageType.PUBLISH,false,MqttQoS.AT_LEAST_ONCE,false,0);MqttPublishVariableHeadervariableHeader=newMqttPublishVariableHeader(topic,0);ByteBufpayload=Unpooled.wrappedBuffer(command.toJSONString().getBytes());MqttPublishMessagemsg=newMqttPublishMessage(mqttFixedHeader,variableHeader,payload);mqttBroker.internalPublish(msg,clientId);System.out.println("✅ 已下发指令到设备 "+username+" → "+command);}catch(Exceptione){e.printStackTrace();}}}

四、测试

使用 MQTTX 进行测试

/** * 测试Controller */@RestController@RequestMapping("/test")publicclassTestControllerextendsBaseController{@AutowiredprivateMqttCommandServicemqttCommandService;/** * MQTT消息发送 */@PostMapping("/command")publicRcommand(@RequestBodyJSONObjectparams){mqttCommandService.sendCommand(params.getString("clientId"),params.getString("username"),params.getJSONObject("command"));returnsuccess();}}