我们是由枫哥组建的IT技术团队成立于2017年致力于帮助IT从业者提供实力成功入职理想企业我们提供一对一学习辅导由知名大厂导师指导分享Java技术、参与项目实战等服务并为学员定制职业规划全面提升竞争力过去8年我们已成功帮助数千名求职者拿到满意的Offerqiuzhiwang.vip。一、为什么选 MQTT 而非 WebSocket维度WebSocketMQTT协议层级传输层裸 TCP应用层基于 TCP/IP消息模式点对点发布/订阅多对多QoS 保障无内置机制0/1/2 三级消息质量带宽消耗较高需自定义心跳极低2 字节固定头离线消息需自行实现内置保留消息Retained适用场景高频双向流游戏、交易IoT、即时通讯、推送MQTT 核心优势以极小代码量和带宽实现可靠的发布/订阅消息服务。二、MQTT 核心概念概念说明类比Publisher消息发布者发送方Subscriber消息订阅者接收方Broker消息代理RabbitMQ/EMQ X邮局Topic主题层级结构如chat/room/123收件地址QoS消息质量等级快递服务等级QoS 等级详解等级语义场景代价0At most once至多一次传感器数据允许丢失最低带宽1At least once至少一次通知推送允许重复一次确认2Exactly once仅一次支付指令不可丢失/重复四次握手三、RabbitMQ 启用 MQTT 服务3.1 Docker 部署# 暴露端口5672(AMQP) 15672(管理) 1883(MQTT) 15675(WebSocket-MQTT)dockerrun-d--namerabbitmq-mqtt\-p5672:5672\-p15672:15672\-p1883:1883\-p15675:15675\-v/mydata/rabbitmq:/var/lib/rabbitmq\rabbitmq:3.12-management3.2 启用插件dockerexec-itrabbitmq-mqttbashrabbitmq-pluginsenablerabbitmq_mqtt rabbitmq_web_mqtt# 输出The following plugins have been configured: rabbitmq_mqtt, rabbitmq_web_mqtt服务验证管理界面http://localhost:15672guest/guestWebSocket 端口15675四、前端纯实现零后端即时通讯4.1 技术栈MQTT.js浏览器 MQTT 客户端库WebSocket 连接ws://host:15675/ws4.2 完整代码!DOCTYPEhtmlhtmllangzh-CNheadmetacharsetUTF-8titleMQTT 即时通讯/titlestyle#messageDiv{height:300px;overflow-y:auto;border:1px solid #ccc;padding:10px;}.msg{margin:5px 0;padding:8px;background:#f0f0f0;border-radius:4px;}.msg.self{background:#d4edda;text-align:right;}/style/headbodyh3当前主题:spanidcurrentTopic/span/h3divinputidtargetTopicplaceholder目标主题valuetestTopicBinputidmessageplaceholder输入消息buttononclicksend()发送/buttonbuttononclickclearMsg()清空/button/divdividmessageDiv/divscriptsrchttps://unpkg.com/mqtt/dist/mqtt.esm.js/scriptscripttypemoduleimportmqttfromhttps://unpkg.com/mqtt/dist/mqtt.esm.js;constWS_URLws://localhost:15675/ws;consttopicnewURLSearchParams(location.search).get(topic)||testTopicA;document.getElementById(currentTopic).textContenttopic;// 连接配置constclientmqtt.connect(WS_URL,{clientId:web_Math.random().toString(16).substr(2,8),clean:true,connectTimeout:4000,reconnectPeriod:1000,});// 连接成功client.on(connect,(){client.subscribe(topic,{qos:1},(err){if(!err)addMsg(✅ 已订阅主题:${topic},system);});});// 接收消息client.on(message,(recvTopic,payload){constmsgpayload.toString();constisSelfrecvTopic!topic;// 简化判断addMsg(${recvTopic}:${msg},isSelf?other:self);});// 发送消息window.send(){consttargetdocument.getElementById(targetTopic).value;constmsgdocument.getElementById(message).value;if(!target||!msg)return;client.publish(target,msg,{qos:1,retain:false});addMsg( 发送至${target}:${msg},self);document.getElementById(message).value;};// UI 辅助functionaddMsg(text,type){constdivdocument.createElement(div);div.classNamemsg${type};div.textContent[${newDate().toLocaleTimeString()}]${text};document.getElementById(messageDiv).appendChild(div);}window.clearMsg(){document.getElementById(messageDiv).innerHTML;};/script/body/html4.3 测试验证窗口 Ahttp://localhost:8080/?topicroomA窗口 Bhttp://localhost:8080/?topicroomB交互流程A 订阅roomA向roomB发送消息B 实时接收双向验证五、Spring Boot 集成服务端介入场景当需要服务端处理消息如存储、过滤、转发时引入 Spring Integration MQTT。5.1 依赖配置dependencygroupIdorg.springframework.integration/groupIdartifactIdspring-integration-mqtt/artifactId/dependencymqtt:broker-url:tcp://localhost:1883client-id:spring-server-${random.uuid}username:guestpassword:guestdefault-topic:broadcastqos:1completion-timeout:50005.2 配置类实现ConfigurationEnableIntegrationpublicclassMqttConfig{BeanpublicMqttPahoClientFactorymqttClientFactory(Value(${mqtt.broker-url})StringbrokerUrl,Value(${mqtt.username})Stringusername,Value(${mqtt.password})Stringpassword){DefaultMqttPahoClientFactoryfactorynewDefaultMqttPahoClientFactory();MqttConnectOptionsoptionsnewMqttConnectOptions();options.setServerURIs(newString[]{brokerUrl});options.setUserName(username);options.setPassword(password.toCharArray());options.setAutomaticReconnect(true);options.setCleanSession(false);// 持久会话factory.setConnectionOptions(options);returnfactory;}}5.3 消息订阅InboundConfigurationpublicclassMqttInboundConfig{BeanpublicMessageChannelmqttInputChannel(){returnnewDirectChannel();}BeanpublicMessageProducerinbound(MqttPahoClientFactoryfactory,Value(${mqtt.default-topic})Stringtopic){MqttPahoMessageDrivenChannelAdapteradapternewMqttPahoMessageDrivenChannelAdapter(spring-subscriber,factory,topic,chat/#,// 通配符订阅notify//urgent// 多级通配);adapter.setCompletionTimeout(5000);adapter.setQos(1);adapter.setConverter(newDefaultPahoMessageConverter());adapter.setOutputChannel(mqttInputChannel());returnadapter;}BeanServiceActivator(inputChannelmqttInputChannel)publicMessageHandlermessageHandler(){returnmessage-{Stringpayloadmessage.getPayload().toString();Stringtopicmessage.getHeaders().get(MqttHeaders.TOPIC,String.class);log.info(收到消息 [topic{}]: {},topic,payload);// 业务处理存储、过滤、转发if(topic.startsWith(chat/)){processChatMessage(topic,payload);}};}privatevoidprocessChatMessage(Stringtopic,Stringpayload){// 持久化到数据库// 敏感词过滤// 推送到其他在线用户}}5.4 消息发布OutboundConfigurationpublicclassMqttOutboundConfig{BeanServiceActivator(inputChannelmqttOutboundChannel)publicMessageHandlermqttOutbound(MqttPahoClientFactoryfactory){MqttPahoMessageHandlerhandlernewMqttPahoMessageHandler(spring-publisher,factory);handler.setAsync(true);// 异步发送handler.setDefaultTopic(broadcast);handler.setDefaultQos(1);returnhandler;}BeanpublicMessageChannelmqttOutboundChannel(){returnnewDirectChannel();}}5.5 网关接口简化调用ComponentMessagingGateway(defaultRequestChannelmqttOutboundChannel)publicinterfaceMqttGateway{voidsend(Stringpayload);voidsend(Stringpayload,Header(MqttHeaders.TOPIC)Stringtopic);voidsend(Stringpayload,Header(MqttHeaders.TOPIC)Stringtopic,Header(MqttHeaders.QOS)intqos);}5.6 REST 控制器RestControllerRequestMapping(/api/mqtt)RequiredArgsConstructorpublicclassMqttController{privatefinalMqttGatewaygateway;PostMapping(/broadcast)publicResponseEntityVoidbroadcast(RequestBodyStringmessage){gateway.send(message);returnResponseEntity.ok().build();}PostMapping(/send/{topic})publicResponseEntityVoidsendToTopic(PathVariableStringtopic,RequestBodyStringmessage,RequestParam(defaultValue1)intqos){gateway.send(message,topic,qos);returnResponseEntity.ok().build();}}六、架构模式对比模式架构适用场景复杂度纯前端浏览器 ↔ RabbitMQ简单聊天、通知⭐Spring 集成浏览器 ↔ RabbitMQ ↔ Spring需要服务端处理⭐⭐⭐混合模式前端 MQTT 后端 WebSocket复杂业务逻辑⭐⭐⭐⭐七、生产环境 checklist检查项配置说明TLS 加密ssl:///wss://生产必开防止窃听认证鉴权RabbitMQ 用户权限按主题隔离权限心跳保活keepAliveInterval检测僵尸连接遗嘱消息willMessage离线通知消息持久化retain: true新订阅者获取最后状态连接限流RabbitMQ 配置防止 DDoS八、总结场景推荐方案快速原型/简单 IM前端直连 MQTT需消息存储/过滤Spring Boot MQTT超大规模百万连接专用 MQTT BrokerEMQ X 集群⭐️推荐:IT枫斗者介绍Java 面试 后端通用面试八股文Java后端企业级实战面试Java后端校招算法学习