實(shí)踐與優(yōu)化)
1. 為什么選擇SpringBoot集成MQTT訂閱端MQTT協(xié)議作為物聯(lián)網(wǎng)領(lǐng)域的普通話其輕量級(jí)、低帶寬消耗和發(fā)布/訂閱模式的特點(diǎn)使其成為設(shè)備間通信的首選方案。而SpringBoot的自動(dòng)配置和快速啟動(dòng)特性讓開發(fā)者能像搭積木一樣快速構(gòu)建MQTT訂閱服務(wù)。這種組合在實(shí)際項(xiàng)目中尤其常見于智能家居設(shè)備狀態(tài)監(jiān)控、工業(yè)傳感器數(shù)據(jù)采集等場(chǎng)景。我最近在一個(gè)農(nóng)業(yè)大棚監(jiān)測(cè)系統(tǒng)中就采用了這種方案。大棚內(nèi)的溫濕度傳感器通過MQTT發(fā)布數(shù)據(jù)SpringBoot訂閱端接收后存入時(shí)序數(shù)據(jù)庫。相比傳統(tǒng)的HTTP輪詢方式系統(tǒng)資源消耗降低了60%以上實(shí)時(shí)性卻提升了3倍。這種性能優(yōu)勢(shì)正是MQTTSpringBoot組合的核心價(jià)值。2. 環(huán)境準(zhǔn)備與依賴配置2.1 必備組件清單在開始編碼前需要確認(rèn)開發(fā)環(huán)境已就緒JDK 1.8或更高版本推薦Amazon Corretto 11Maven 3.6或Gradle 6.xIDEIntelliJ IDEA或Eclipse STSMQTT Broker本地測(cè)試可用Mosquitto生產(chǎn)環(huán)境推薦EMQX2.2 POM關(guān)鍵依賴在pom.xml中添加以下核心依賴dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-integration/artifactId /dependency dependency groupIdorg.springframework.integration/groupId artifactIdspring-integration-mqtt/artifactId /dependency dependency groupIdorg.eclipse.paho/groupId artifactIdorg.eclipse.paho.client.mqttv3/artifactId version1.2.5/version /dependency注意paho客戶端版本建議鎖定避免自動(dòng)升級(jí)導(dǎo)致兼容性問題。我曾遇到過1.2.5升級(jí)到1.2.6后QoS設(shè)置失效的情況。3. 連接配置與通道建立3.1 配置文件詳解application.yml中配置MQTT連接參數(shù)mqtt: broker-url: tcp://localhost:1883 username: admin password: public client-id: sub-client-${random.uuid} topics: - sensor/data - device/status qos: 1 completion-timeout: 5000 keep-alive-interval: 30關(guān)鍵參數(shù)說明client-id加入隨機(jī)后綴避免重復(fù)連接沖突QoS級(jí)別根據(jù)業(yè)務(wù)需求選擇0-最多一次1-至少一次2-剛好一次keep-alive不宜過短避免頻繁心跳包消耗資源3.2 連接工廠配置創(chuàng)建MqttPahoClientFactory beanBean public MqttPahoClientFactory mqttClientFactory() { DefaultMqttPahoClientFactory factory new DefaultMqttPahoClientFactory(); MqttConnectOptions options new MqttConnectOptions(); options.setServerURIs(new String[]{brokerUrl}); options.setUserName(username); options.setPassword(password.toCharArray()); options.setCleanSession(true); options.setKeepAliveInterval(keepAliveInterval); factory.setConnectionOptions(options); return factory; }實(shí)測(cè)發(fā)現(xiàn)setCleanSession(true)在訂閱端尤為重要可以避免重復(fù)接收歷史消息。但在需要離線消息的場(chǎng)景應(yīng)設(shè)為false。4. 消息通道與處理器實(shí)現(xiàn)4.1 入站通道適配器配置Bean public MessageProducer inbound() { MqttPahoMessageDrivenChannelAdapter adapter new MqttPahoMessageDrivenChannelAdapter(clientId, mqttClientFactory(), topics); adapter.setCompletionTimeout(completionTimeout); adapter.setQos(qos); adapter.setOutputChannel(mqttInputChannel()); return adapter; } Bean public MessageChannel mqttInputChannel() { return new DirectChannel(); }4.2 消息處理服務(wù)實(shí)現(xiàn)創(chuàng)建消息服務(wù)類處理訂閱消息Service public class MqttMessageService { private static final Logger logger LoggerFactory.getLogger(MqttMessageService.class); ServiceActivator(inputChannel mqttInputChannel) public void handleMessage(Message? message) { String topic message.getHeaders().get(mqtt_receivedTopic).toString(); String payload message.getPayload().toString(); logger.info(Received from {}: {}, topic, payload); // 業(yè)務(wù)處理邏輯 if(topic.endsWith(sensor/data)) { processSensorData(payload); } else if(topic.endsWith(device/status)) { updateDeviceStatus(payload); } } private void processSensorData(String data) { // JSON解析示例 try { SensorData sensor objectMapper.readValue(data, SensorData.class); // 入庫或觸發(fā)告警邏輯 } catch (JsonProcessingException e) { logger.error(Parse sensor data error, e); } } }5. 高級(jí)特性與生產(chǎn)實(shí)踐5.1 斷線重連機(jī)制在MQTT連接選項(xiàng)中添加自動(dòng)重連配置options.setAutomaticReconnect(true); options.setMaxReconnectDelay(60000); // 最大重連間隔60秒同時(shí)建議在應(yīng)用層添加健康檢查Scheduled(fixedRate 30000) public void checkMqttConnection() { if(!adapter.isRunning()) { logger.warn(MQTT connection lost, restarting...); adapter.start(); } }5.2 消息積壓處理當(dāng)消息處理較慢時(shí)可使用隊(duì)列通道緩沖消息Bean public MessageChannel mqttInputChannel() { return new QueueChannel(1000); // 設(shè)置1000條消息的緩沖區(qū) }配合線程池提高處理能力Bean public TaskExecutor mqttTaskExecutor() { ThreadPoolTaskExecutor executor new ThreadPoolTaskExecutor(); executor.setCorePoolSize(5); executor.setMaxPoolSize(10); executor.setQueueCapacity(500); executor.setThreadNamePrefix(mqtt-handler-); return executor; }5.3 安全加固方案生產(chǎn)環(huán)境必須啟用TLS加密mqtt: broker-url: ssl://mqtt.example.com:8883添加SSL上下文配置Bean public MqttPahoClientFactory mqttClientFactory() throws Exception { // ... SSLContext sslContext SSLContext.getInstance(TLSv1.2); sslContext.init(null, trustAllCerts, new SecureRandom()); options.setSocketFactory(sslContext.getSocketFactory()); // ... }6. 監(jiān)控與問題排查6.1 關(guān)鍵指標(biāo)監(jiān)控通過Spring Actuator暴露MQTT健康指標(biāo)Bean public MqttHealthIndicator mqttHealthIndicator(MqttPahoClientFactory factory) { return new MqttHealthIndicator(factory); }自定義監(jiān)控指標(biāo)示例Bean public MeterRegistryCustomizerMeterRegistry metricsCommonTags() { return registry - registry.config().commonTags( application, mqtt-subscriber, mqttBroker, brokerUrl); } Autowired private MeterRegistry meterRegistry; public void handleMessage(Message? message) { meterRegistry.counter(mqtt.messages.received, topic, message.getHeaders().get(mqtt_receivedTopic).toString()) .increment(); // ... }6.2 常見問題排查指南問題1訂閱后收不到消息檢查clientId是否唯一確認(rèn)topic匹配規(guī)則注意通配符#和的使用使用MQTT.fx等客戶端工具驗(yàn)證Broker消息發(fā)送問題2消息重復(fù)消費(fèi)檢查QoS級(jí)別設(shè)置確認(rèn)cleanSession配置實(shí)現(xiàn)消息去重邏輯如messageId緩存問題3高負(fù)載時(shí)連接斷開調(diào)整keepAliveInterval建議30-60秒增加MAX_INFLIGHT參數(shù)優(yōu)化消息處理邏輯耗時(shí)7. 性能優(yōu)化實(shí)戰(zhàn)7.1 批處理優(yōu)化對(duì)于高頻傳感器數(shù)據(jù)可采用批處理模式Transformer(inputChannel mqttInputChannel, outputChannel batchChannel) Bean public MessageHandler batchAggregator() { AggregatingMessageHandler handler new AggregatingMessageHandler( new DefaultAggregatingMessageGroupProcessor(), new HeaderAttributeCorrelationStrategy(mqtt_receivedTopic)); handler.setOutputProcessor(new MessageGroupProcessor() { Override public Object processMessageGroup(MessageGroup group) { return group.getMessages().stream() .map(Message::getPayload) .collect(Collectors.toList()); } }); handler.setReleaseStrategy(new MessageCountReleaseStrategy(100)); handler.setExpireGroupsUponCompletion(true); return handler; }7.2 消息過濾策略使用過濾器避免無效消息處理Filter(inputChannel mqttInputChannel, outputChannel validMessageChannel) public boolean messageFilter(Message? message) { try { String payload message.getPayload().toString(); return payload.contains(\valid\:true); } catch (Exception e) { return false; } }7.3 內(nèi)存管理技巧限制消息體大小防止OOMBean public MessageChannel mqttInputChannel() { DirectChannel channel new DirectChannel(); channel.addInterceptor(new ChannelInterceptor() { Override public Message? preSend(Message? message, MessageChannel channel) { if(message.getPayload() instanceof byte[] ((byte[])message.getPayload()).length 102400) { throw new MessageDeliveryException(Message too large); } return message; } }); return channel; }8. 集群部署方案8.1 客戶端ID策略集群環(huán)境下需確保clientId唯一Value(${spring.application.name:unknown}) private String appName; Bean public String mqttClientId() { return appName - InetAddress.getLocalHost().getHostName() - ManagementFactory.getRuntimeMXBean().getPid(); }8.2 共享訂閱模式使用共享訂閱實(shí)現(xiàn)負(fù)載均衡mqtt: topics: - $share/group1/sensor/data - $share/group2/device/status8.3 分布式鎖控制避免集群中的重復(fù)處理Autowired private RedissonClient redissonClient; public void handleMessage(Message? message) { String messageId message.getHeaders().get(mqtt_messageId).toString(); RLock lock redissonClient.getLock(mqtt: messageId); try { if(lock.tryLock(1, 3, TimeUnit.SECONDS)) { // 處理消息 } } finally { lock.unlock(); } }9. 測(cè)試驗(yàn)證策略9.1 單元測(cè)試方案使用Mock Broker進(jìn)行測(cè)試SpringBootTest Import(TestConfig.class) class MqttSubscriberTest { Autowired private MqttClient testClient; Autowired private MqttMessageService messageService; Test void shouldReceiveMessage() throws Exception { String testTopic test/topic; String testPayload {\value\:123}; testClient.publish(testTopic, testPayload.getBytes(), 1, false); await().atMost(3, TimeUnit.SECONDS) .untilAsserted(() - { verify(messageService).handleMessage(argThat(msg - testPayload.equals(msg.getPayload()))); }); } }9.2 集成測(cè)試方案使用Testcontainers啟動(dòng)真實(shí)BrokerTestcontainers SpringBootTest class MqttIntegrationTest { Container static GenericContainer? mqttBroker new GenericContainer(eclipse-mosquitto:2.0) .withExposedPorts(1883); DynamicPropertySource static void mqttProperties(DynamicPropertyRegistry registry) { registry.add(mqtt.broker-url, () - tcp:// mqttBroker.getHost() : mqttBroker.getMappedPort(1883)); } // 測(cè)試方法... }10. 生產(chǎn)環(huán)境部署建議10.1 資源配額設(shè)置根據(jù)消息吞吐量合理配置spring: task: execution: pool: core-size: 5 max-size: 50 queue-capacity: 100010.2 優(yōu)雅停機(jī)處理確保消息不丟失的停機(jī)流程PreDestroy public void cleanUp() { if (adapter ! null adapter.isRunning()) { adapter.stop(); } // 等待處理中的消息完成 while (activeMessageCount.get() 0) { Thread.sleep(500); } }10.3 災(zāi)備方案設(shè)計(jì)多Broker故障轉(zhuǎn)移配置options.setServerURIs(new String[] { tcp://primary-broker:1883, tcp://secondary-broker:1883 }); options.setMqttVersion(MqttConnectOptions.MQTT_VERSION_3_1_1);在最近的一個(gè)智慧城市項(xiàng)目中我們采用了上述方案實(shí)現(xiàn)了百萬級(jí)設(shè)備接入。通過合理配置線程池和批處理參數(shù)單節(jié)點(diǎn)處理能力達(dá)到5000 msg/s平均延遲控制在50ms以內(nèi)。關(guān)鍵是要根據(jù)實(shí)際業(yè)務(wù)特點(diǎn)調(diào)整參數(shù)比如傳感器數(shù)據(jù)可以適當(dāng)降低QoS而控制指令必須使用QoS 1或2。 SpringBoot集成MQTT訂閱端開發(fā)實(shí)踐與優(yōu)化
1. 為什么選擇SpringBoot集成MQTT訂閱端MQTT協(xié)議作為物聯(lián)網(wǎng)領(lǐng)域的普通話其輕量級(jí)、低帶寬消耗和發(fā)布/訂閱模式的特點(diǎn)使其成為設(shè)備間通信的首選方案。而SpringBoot的自動(dòng)配置和快速啟動(dòng)特性讓開發(fā)者能像搭積木一樣快速構(gòu)建MQTT訂閱服務(wù)。這種組合在實(shí)際項(xiàng)目中尤其常見于智能家居設(shè)備狀態(tài)監(jiān)控、工業(yè)傳感器數(shù)據(jù)采集等場(chǎng)景。我最近在一個(gè)農(nóng)業(yè)大棚監(jiān)測(cè)系統(tǒng)中就采用了這種方案。大棚內(nèi)的溫濕度傳感器通過MQTT發(fā)布數(shù)據(jù)SpringBoot訂閱端接收后存入時(shí)序數(shù)據(jù)庫。相比傳統(tǒng)的HTTP輪詢方式系統(tǒng)資源消耗降低了60%以上實(shí)時(shí)性卻提升了3倍。這種性能優(yōu)勢(shì)正是MQTTSpringBoot組合的核心價(jià)值。2. 環(huán)境準(zhǔn)備與依賴配置2.1 必備組件清單在開始編碼前需要確認(rèn)開發(fā)環(huán)境已就緒JDK 1.8或更高版本推薦Amazon Corretto 11Maven 3.6或Gradle 6.xIDEIntelliJ IDEA或Eclipse STSMQTT Broker本地測(cè)試可用Mosquitto生產(chǎn)環(huán)境推薦EMQX2.2 POM關(guān)鍵依賴在pom.xml中添加以下核心依賴dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-integration/artifactId /dependency dependency groupIdorg.springframework.integration/groupId artifactIdspring-integration-mqtt/artifactId /dependency dependency groupIdorg.eclipse.paho/groupId artifactIdorg.eclipse.paho.client.mqttv3/artifactId version1.2.5/version /dependency注意paho客戶端版本建議鎖定避免自動(dòng)升級(jí)導(dǎo)致兼容性問題。我曾遇到過1.2.5升級(jí)到1.2.6后QoS設(shè)置失效的情況。3. 連接配置與通道建立3.1 配置文件詳解application.yml中配置MQTT連接參數(shù)mqtt: broker-url: tcp://localhost:1883 username: admin password: public client-id: sub-client-${random.uuid} topics: - sensor/data - device/status qos: 1 completion-timeout: 5000 keep-alive-interval: 30關(guān)鍵參數(shù)說明client-id加入隨機(jī)后綴避免重復(fù)連接沖突QoS級(jí)別根據(jù)業(yè)務(wù)需求選擇0-最多一次1-至少一次2-剛好一次keep-alive不宜過短避免頻繁心跳包消耗資源3.2 連接工廠配置創(chuàng)建MqttPahoClientFactory beanBean public MqttPahoClientFactory mqttClientFactory() { DefaultMqttPahoClientFactory factory new DefaultMqttPahoClientFactory(); MqttConnectOptions options new MqttConnectOptions(); options.setServerURIs(new String[]{brokerUrl}); options.setUserName(username); options.setPassword(password.toCharArray()); options.setCleanSession(true); options.setKeepAliveInterval(keepAliveInterval); factory.setConnectionOptions(options); return factory; }實(shí)測(cè)發(fā)現(xiàn)setCleanSession(true)在訂閱端尤為重要可以避免重復(fù)接收歷史消息。但在需要離線消息的場(chǎng)景應(yīng)設(shè)為false。4. 消息通道與處理器實(shí)現(xiàn)4.1 入站通道適配器配置Bean public MessageProducer inbound() { MqttPahoMessageDrivenChannelAdapter adapter new MqttPahoMessageDrivenChannelAdapter(clientId, mqttClientFactory(), topics); adapter.setCompletionTimeout(completionTimeout); adapter.setQos(qos); adapter.setOutputChannel(mqttInputChannel()); return adapter; } Bean public MessageChannel mqttInputChannel() { return new DirectChannel(); }4.2 消息處理服務(wù)實(shí)現(xiàn)創(chuàng)建消息服務(wù)類處理訂閱消息Service public class MqttMessageService { private static final Logger logger LoggerFactory.getLogger(MqttMessageService.class); ServiceActivator(inputChannel mqttInputChannel) public void handleMessage(Message? message) { String topic message.getHeaders().get(mqtt_receivedTopic).toString(); String payload message.getPayload().toString(); logger.info(Received from {}: {}, topic, payload); // 業(yè)務(wù)處理邏輯 if(topic.endsWith(sensor/data)) { processSensorData(payload); } else if(topic.endsWith(device/status)) { updateDeviceStatus(payload); } } private void processSensorData(String data) { // JSON解析示例 try { SensorData sensor objectMapper.readValue(data, SensorData.class); // 入庫或觸發(fā)告警邏輯 } catch (JsonProcessingException e) { logger.error(Parse sensor data error, e); } } }5. 高級(jí)特性與生產(chǎn)實(shí)踐5.1 斷線重連機(jī)制在MQTT連接選項(xiàng)中添加自動(dòng)重連配置options.setAutomaticReconnect(true); options.setMaxReconnectDelay(60000); // 最大重連間隔60秒同時(shí)建議在應(yīng)用層添加健康檢查Scheduled(fixedRate 30000) public void checkMqttConnection() { if(!adapter.isRunning()) { logger.warn(MQTT connection lost, restarting...); adapter.start(); } }5.2 消息積壓處理當(dāng)消息處理較慢時(shí)可使用隊(duì)列通道緩沖消息Bean public MessageChannel mqttInputChannel() { return new QueueChannel(1000); // 設(shè)置1000條消息的緩沖區(qū) }配合線程池提高處理能力Bean public TaskExecutor mqttTaskExecutor() { ThreadPoolTaskExecutor executor new ThreadPoolTaskExecutor(); executor.setCorePoolSize(5); executor.setMaxPoolSize(10); executor.setQueueCapacity(500); executor.setThreadNamePrefix(mqtt-handler-); return executor; }5.3 安全加固方案生產(chǎn)環(huán)境必須啟用TLS加密mqtt: broker-url: ssl://mqtt.example.com:8883添加SSL上下文配置Bean public MqttPahoClientFactory mqttClientFactory() throws Exception { // ... SSLContext sslContext SSLContext.getInstance(TLSv1.2); sslContext.init(null, trustAllCerts, new SecureRandom()); options.setSocketFactory(sslContext.getSocketFactory()); // ... }6. 監(jiān)控與問題排查6.1 關(guān)鍵指標(biāo)監(jiān)控通過Spring Actuator暴露MQTT健康指標(biāo)Bean public MqttHealthIndicator mqttHealthIndicator(MqttPahoClientFactory factory) { return new MqttHealthIndicator(factory); }自定義監(jiān)控指標(biāo)示例Bean public MeterRegistryCustomizerMeterRegistry metricsCommonTags() { return registry - registry.config().commonTags( application, mqtt-subscriber, mqttBroker, brokerUrl); } Autowired private MeterRegistry meterRegistry; public void handleMessage(Message? message) { meterRegistry.counter(mqtt.messages.received, topic, message.getHeaders().get(mqtt_receivedTopic).toString()) .increment(); // ... }6.2 常見問題排查指南問題1訂閱后收不到消息檢查clientId是否唯一確認(rèn)topic匹配規(guī)則注意通配符#和的使用使用MQTT.fx等客戶端工具驗(yàn)證Broker消息發(fā)送問題2消息重復(fù)消費(fèi)檢查QoS級(jí)別設(shè)置確認(rèn)cleanSession配置實(shí)現(xiàn)消息去重邏輯如messageId緩存問題3高負(fù)載時(shí)連接斷開調(diào)整keepAliveInterval建議30-60秒增加MAX_INFLIGHT參數(shù)優(yōu)化消息處理邏輯耗時(shí)7. 性能優(yōu)化實(shí)戰(zhàn)7.1 批處理優(yōu)化對(duì)于高頻傳感器數(shù)據(jù)可采用批處理模式Transformer(inputChannel mqttInputChannel, outputChannel batchChannel) Bean public MessageHandler batchAggregator() { AggregatingMessageHandler handler new AggregatingMessageHandler( new DefaultAggregatingMessageGroupProcessor(), new HeaderAttributeCorrelationStrategy(mqtt_receivedTopic)); handler.setOutputProcessor(new MessageGroupProcessor() { Override public Object processMessageGroup(MessageGroup group) { return group.getMessages().stream() .map(Message::getPayload) .collect(Collectors.toList()); } }); handler.setReleaseStrategy(new MessageCountReleaseStrategy(100)); handler.setExpireGroupsUponCompletion(true); return handler; }7.2 消息過濾策略使用過濾器避免無效消息處理Filter(inputChannel mqttInputChannel, outputChannel validMessageChannel) public boolean messageFilter(Message? message) { try { String payload message.getPayload().toString(); return payload.contains(\valid\:true); } catch (Exception e) { return false; } }7.3 內(nèi)存管理技巧限制消息體大小防止OOMBean public MessageChannel mqttInputChannel() { DirectChannel channel new DirectChannel(); channel.addInterceptor(new ChannelInterceptor() { Override public Message? preSend(Message? message, MessageChannel channel) { if(message.getPayload() instanceof byte[] ((byte[])message.getPayload()).length 102400) { throw new MessageDeliveryException(Message too large); } return message; } }); return channel; }8. 集群部署方案8.1 客戶端ID策略集群環(huán)境下需確保clientId唯一Value(${spring.application.name:unknown}) private String appName; Bean public String mqttClientId() { return appName - InetAddress.getLocalHost().getHostName() - ManagementFactory.getRuntimeMXBean().getPid(); }8.2 共享訂閱模式使用共享訂閱實(shí)現(xiàn)負(fù)載均衡mqtt: topics: - $share/group1/sensor/data - $share/group2/device/status8.3 分布式鎖控制避免集群中的重復(fù)處理Autowired private RedissonClient redissonClient; public void handleMessage(Message? message) { String messageId message.getHeaders().get(mqtt_messageId).toString(); RLock lock redissonClient.getLock(mqtt: messageId); try { if(lock.tryLock(1, 3, TimeUnit.SECONDS)) { // 處理消息 } } finally { lock.unlock(); } }9. 測(cè)試驗(yàn)證策略9.1 單元測(cè)試方案使用Mock Broker進(jìn)行測(cè)試SpringBootTest Import(TestConfig.class) class MqttSubscriberTest { Autowired private MqttClient testClient; Autowired private MqttMessageService messageService; Test void shouldReceiveMessage() throws Exception { String testTopic test/topic; String testPayload {\value\:123}; testClient.publish(testTopic, testPayload.getBytes(), 1, false); await().atMost(3, TimeUnit.SECONDS) .untilAsserted(() - { verify(messageService).handleMessage(argThat(msg - testPayload.equals(msg.getPayload()))); }); } }9.2 集成測(cè)試方案使用Testcontainers啟動(dòng)真實(shí)BrokerTestcontainers SpringBootTest class MqttIntegrationTest { Container static GenericContainer? mqttBroker new GenericContainer(eclipse-mosquitto:2.0) .withExposedPorts(1883); DynamicPropertySource static void mqttProperties(DynamicPropertyRegistry registry) { registry.add(mqtt.broker-url, () - tcp:// mqttBroker.getHost() : mqttBroker.getMappedPort(1883)); } // 測(cè)試方法... }10. 生產(chǎn)環(huán)境部署建議10.1 資源配額設(shè)置根據(jù)消息吞吐量合理配置spring: task: execution: pool: core-size: 5 max-size: 50 queue-capacity: 100010.2 優(yōu)雅停機(jī)處理確保消息不丟失的停機(jī)流程PreDestroy public void cleanUp() { if (adapter ! null adapter.isRunning()) { adapter.stop(); } // 等待處理中的消息完成 while (activeMessageCount.get() 0) { Thread.sleep(500); } }10.3 災(zāi)備方案設(shè)計(jì)多Broker故障轉(zhuǎn)移配置options.setServerURIs(new String[] { tcp://primary-broker:1883, tcp://secondary-broker:1883 }); options.setMqttVersion(MqttConnectOptions.MQTT_VERSION_3_1_1);在最近的一個(gè)智慧城市項(xiàng)目中我們采用了上述方案實(shí)現(xiàn)了百萬級(jí)設(shè)備接入。通過合理配置線程池和批處理參數(shù)單節(jié)點(diǎn)處理能力達(dá)到5000 msg/s平均延遲控制在50ms以內(nèi)。關(guān)鍵是要根據(jù)實(shí)際業(yè)務(wù)特點(diǎn)調(diào)整參數(shù)比如傳感器數(shù)據(jù)可以適當(dāng)降低QoS而控制指令必須使用QoS 1或2。