Angular应用中接收MQTT消息失败，寻求解决方案

问题背景 #

按照ngx-mqtt文档实现代码后，Angular应用无法接收MQTT消息。相关实现代码及发送消息的操作截图如下：

app.module.ts 代码 #

import {
  IMqttMessage,
  MqttModule,
  IMqttServiceOptions,
  MqttService
} from 'ngx-mqtt';

export const MQTT_SERVICE_OPTIONS: IMqttServiceOptions = {
  hostname: 'localhost',
  port: 9001,
  path: '/mqtt'
};

app.component.ts 代码 #

import { Subscription } from 'rxjs';
import { MqttService, IMqttMessage } from 'ngx-mqtt';
import { NgxSerial } from 'ngx-serial';

export class AppComponent {
  subscription: Subscription | undefined;
  message: string = '';
  serial: NgxSerial;

  constructor(private mqttService: MqttService) {
    this.serial = new NgxSerial(this.dataHandler);
    console.log(mqttService);
    this.subscription = this.mqttService.observe('my/topic').subscribe((message: IMqttMessage) => {
      this.message = message.payload.toString();
      console.log('Received message:', message.payload.toString());
    });
  }

  public unsafePublish(topic: string, message: string): void {
    this.mqttService.unsafePublish(topic, message, { qos: 1, retain: true });
  }

  public ngOnDestroy() {
    if (this.subscription) {
      this.subscription.unsubscribe();
    }
  }

  private dataHandler = (data: any) => {
    // 串口数据处理逻辑
  };
}

发送消息操作截图：

排查与解决步骤 #

1. 确认MQTT连接状态 #

添加连接状态监听，排查是否成功建立连接：

constructor(private mqttService: MqttService) {
  // ...原有代码
  this.mqttService.onConnect.subscribe(() => console.log('MQTT连接成功'));
  this.mqttService.onError.subscribe(err => console.error('MQTT连接错误:', err));
  this.mqttService.onClose.subscribe(() => console.log('MQTT连接关闭'));
}

• 若连接失败，检查：
  • MQTT Broker的WebSocket端口（9001）及路径（/mqtt）是否配置正确
  • 浏览器控制台是否有跨域错误，如有需在Broker中配置跨域允许规则

2. 校验Topic与QoS匹配 #

• 确认发送与订阅的Topic完全一致（MQTT Topic大小写敏感）
• 检查发送消息的QoS等级：代码中订阅默认QoS为0，若发送端使用QoS1/2，需确保Broker支持并正确转发

3. 本地自测发布订阅 #

在组件中调用this.unsafePublish('my/topic', 'test message')，查看控制台是否能收到消息：

• 能收到：说明本地订阅逻辑正常，问题出在外部发送环节（比如外部客户端的连接配置、Topic正确性）
• 不能收到：检查MqttModule是否在AppModule中正确初始化：

  @NgModule({
    imports: [
      // ...其他模块
      MqttModule.forRoot(MQTT_SERVICE_OPTIONS)
    ]
  })
  

4. 检查MQTT Broker配置 #

• 确认Broker已开启WebSocket服务，端口为9001、路径为/mqtt
• 确认Broker允许匿名连接（代码未配置用户名密码时）
• 使用第三方MQTT客户端（如MQTTX）连接ws://localhost:9001/mqtt，订阅my/topic并发送消息，验证Broker是否正常转发

5. 排查RxJS订阅异常 #

• 确认订阅未被提前取消，检查是否有其他逻辑在生命周期中执行了unsubscribe
• 为订阅添加错误回调，排查订阅过程中的异常：

  this.subscription = this.mqttService.observe('my/topic').subscribe({
    next: (message: IMqttMessage) => {
      this.message = message.payload.toString();
      console.log('Received message:', message.payload.toString());
    },
    error: err => console.error('订阅错误:', err)
  });
  

内容的提问来源于stack exchange，提问作者Katerina Maria Gerakianaki