跨平台物联网通信实战：UniApp双端MQTT适配与EMQX深度调优指南

如果你正在用UniApp开发需要实时通信的物联网应用，肯定遇到过这样的场景：H5端跑得好好的MQTT连接，一到微信小程序就各种报错；模拟器调试一切正常，真机测试却死活连不上服务器。这不仅仅是协议差异的问题，更涉及到平台限制、网络策略、安全规范等一系列复杂因素。

我最近在一个智慧农业监控项目中就踩遍了这些坑。项目需要同时在Web管理后台和微信小程序上展示传感器数据，最初以为用同一套MQTT代码就能搞定，结果发现H5用WebSocket连接正常，小程序端却频频失败。经过几天的调试和排查，终于总结出了一套完整的解决方案。今天我就把这些实战经验分享给你，帮你避开我走过的弯路。

1. 理解跨平台MQTT通信的核心差异

在开始编码之前，我们必须先搞清楚不同平台对MQTT连接的限制和要求。这不是简单的"换个协议前缀"那么简单，而是涉及到底层网络实现的根本差异。

1.1 协议栈的微妙区别

H5环境和小程序环境虽然都基于JavaScript，但它们的网络层实现完全不同。H5使用标准的WebSocket API，而微信小程序使用的是自己封装的网络接口。这种差异直接影响了MQTT客户端的连接方式。

H5环境支持标准的WebSocket协议，你可以使用ws://（非加密）或wss://（加密）协议。这意味着你可以直接使用浏览器原生的WebSocket能力，连接过程相对简单。

// H5环境的标准连接方式
const client = mqtt.connect('ws://your-server.com:8083/mqtt', options)

微信小程序环境则有自己的特殊要求。首先，小程序不允许直接使用ws://协议，必须使用wx://（非加密）或wxs://（加密）。其次，小程序要求所有网络请求的域名必须在小程序后台配置的合法域名列表中。

// 小程序环境的连接方式
const client = mqtt.connect('wxs://your-server.com:443/mqtt', options)

1.2 平台限制对比表

为了更清晰地理解这些差异，我整理了一个详细的对比表格：

特性维度	H5环境	微信小程序环境	关键影响
协议支持	ws://, wss://	wx://, wxs://	连接字符串必须正确
域名要求	无限制	必须配置合法域名	真机调试必配
端口限制	无特殊限制	通常限制443端口	影响服务器配置
TLS版本	支持多种TLS版本	要求TLS 1.2+	证书配置要求高
连接超时	浏览器默认设置	小程序有独立超时机制	需要分别优化
后台运行	标签页切换可能断开	有独立生命周期管理	重连策略需调整

注意：微信小程序对TLS证书的要求特别严格。如果你的服务器证书链不完整，或者使用了自签名证书，在开发工具中可能能连接成功，但真机调试一定会失败。这是很多开发者容易忽略的关键点。

1.3 MQTT.js版本选择的艺术

选择合适的MQTT.js版本至关重要。不同版本对小程序环境的支持程度不同，选错了版本可能导致各种诡异的问题。

我在实际项目中测试了多个版本，发现以下规律：

• v2.18.8：兼容性最好，在小程序环境中表现稳定，但功能相对较少
• v3.0.0：功能完整，但在某些小程序真机环境中存在连接问题
• v4.1.0：平衡了功能和兼容性，是目前比较推荐的选择

# 安装推荐版本
npm install mqtt@4.1.0

如果你遇到真机调试连接失败的问题，第一个要检查的就是MQTT.js版本。我建议先使用v4.1.0，如果还有问题再尝试v2.18.8作为备选。

2. 构建智能协议适配层

条件编译是UniApp跨平台开发的核心技术，但仅仅使用#ifdef和#endif还不够。我们需要构建一个更智能、更健壮的协议适配层。

2.1 基础条件编译实现

最基本的适配方案是通过条件编译区分不同平台：

// 基础的条件编译适配
class MQTTAdapter {
  constructor(config) {
    this.config = config
    this.client = null
  }

  connect() {
    let connectionUrl = ''
    
    // #ifdef H5
    connectionUrl = `ws://${this.config.host}:${this.config.port}${this.config.path}`
    // #endif
    
    // #ifdef MP-WEIXIN
    const protocol = this.config.useSSL ? 'wxs' : 'wx'
    connectionUrl = `${protocol}://${this.config.host}:${this.config.port}${this.config.path}`
    // #endif
    
    this.client = mqtt.connect(connectionUrl, this.getOptions())
    this.setupEventHandlers()
  }
  
  getOptions() {
    // 生成唯一的clientId，避免重复连接问题
    const clientId = `mqtt_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`
    
    return {
      clientId,
      username: this.config.username,
      password: this.config.password,
      clean: true,
      reconnectPeriod: 5000,  // 5秒重连间隔
      connectTimeout: 30000,   // 30秒连接超时
      keepalive: 60,          // 60秒心跳间隔
      resubscribe: true       // 自动重新订阅
    }
  }
}

这种方案虽然简单，但存在一个明显的问题：代码重复，维护困难。每次修改连接逻辑都需要在两个地方同步更新。

2.2 进阶：工厂模式适配器

我推荐使用工厂模式来构建更优雅的适配层。这种设计模式将不同平台的连接逻辑封装到独立的适配器中，主代码只需要调用统一的接口。

// 协议适配器工厂
class MQTTAdapterFactory {
  static createAdapter(platform, config) {
    switch(platform) {
      case 'H5':
        return new H5MQTTAdapter(config)
      case 'MP-WEIXIN':
        return new WeChatMQTTAdapter(config)
      case 'APP-PLUS':
        return new AppMQTTAdapter(config)
      default:
        throw new Error(`不支持的平台: ${platform}`)
    }
  }
}

// H5适配器实现
class H5MQTTAdapter extends BaseMQTTAdapter {
  buildConnectionUrl() {
    const protocol = this.config.useSSL ? 'wss' : 'ws'
    return `${protocol}://${this.config.host}:${this.config.port}${this.config.path}`
  }
  
  // H5特有的配置
  getPlatformSpecificOptions() {
    return {
      wsOptions: {
        // H5环境特有的WebSocket配置
        headers: {
          'User-Agent': 'UniApp-H5-MQTT-Client'
        }
      }
    }
  }
}

// 微信小程序适配器实现
class WeChatMQTTAdapter extends BaseMQTTAdapter {
  buildConnectionUrl() {
    const protocol = this.config.useSSL ? 'wxs' : 'wx'
    // 小程序强制要求使用443端口进行SSL连接
    const port = this.config.useSSL ? 443 : this.config.port
    return `${protocol}://${this.config.host}:${port}${this.config.path}`
  }
  
  getPlatformSpecificOptions() {
    return {
      // 小程序环境需要更短的超时时间
      connectTimeout: 15000,
      // 小程序后台运行时的心跳策略
      keepalive: 30
    }
  }
}

2.3 自动平台检测与适配

为了让代码更加智能，我们可以实现自动平台检测功能：

// 自动平台检测适配器
class AutoDetectMQTTAdapter {
  constructor(config) {
    this.config = config
    this.platform = this.detectPlatform()
    this.adapter = null
  }
  
  detectPlatform() {
    // #ifdef H5
    return 'H5'
    // #endif
    
    // #ifdef MP-WEIXIN
    return 'MP-WEIXIN'
    // #endif
    
    // #ifdef APP-PLUS
    return 'APP-PLUS'
    // #endif
    
    // 默认回退到H5
    return 'H5'
  }
  
  connect() {
    this.adapter = MQTTAdapterFactory.createAdapter(this.platform, this.config)
    return this.adapter.connect()
  }
  
  // 代理所有MQTT客户端方法
  subscribe(topic, options) {
    return this.adapter.subscribe(topic, options)
  }
  
  publish(topic, message, options) {
    return this.adapter.publish(topic, message, options)
  }
  
  end() {
    return this.adapter.end()
  }
}

// 使用示例
const mqttClient = new AutoDetectMQTTAdapter({
  host: 'your-mqtt-server.com',
  port: 8083,
  path: '/mqtt',
  useSSL: true,
  username: 'your-username',
  password: 'your-password'
})

// 无需关心当前平台，适配器会自动处理
mqttClient.connect()

这种设计的好处是业务代码完全不需要关心当前运行在哪个平台，所有平台差异都被适配层隐藏了。当需要支持新的平台时，只需要添加新的适配器类即可。

3. EMQX服务器配置与优化

MQTT客户端的适配只是问题的一半，服务器端的配置同样重要。EMQX作为流行的MQTT服务器，有很多配置选项需要特别注意。

3.1 基础EMQX配置

对于UniApp跨平台应用，EMQX需要特别配置WebSocket和SSL支持：

# EMQX配置文件 emqx.conf 的关键配置

# WebSocket监听配置
listeners.ws.default {
  bind = "0.0.0.0:8083"
  max_connections = 102400
  websocket {
    path = "/mqtt"
    compress = true
    idle_timeout = "15s"
  }
}

# SSL WebSocket监听配置（小程序必须）
listeners.wss.default {
  bind = "0.0.0.0:8084"
  max_connections = 102400
  ssl_options {
    keyfile = "/etc/emqx/certs/key.pem"
    certfile = "/etc/emqx/certs/cert.pem"
    cacertfile = "/etc/emqx/certs/cacert.pem"
  }
  websocket {
    path = "/mqtt"
    compress = true
    idle_timeout = "15s"
  }
}

# 客户端限制配置（防止恶意连接）
zone.external {
  max_connections = 10000
  max_packet_size = "10MB"
  idle_timeout = "30s"
  enable_ban = true
  enable_flapping_detect = true
}

3.2 SSL证书配置要点

微信小程序对SSL证书的要求非常严格，以下是必须注意的几点：

1. 证书链必须完整：不能只提供服务器证书，必须包含完整的证书链
2. 支持TLS 1.2+：确保服务器配置了现代TLS版本
3. SAN扩展必须包含域名：证书的主题备用名称必须包含你使用的域名

你可以使用以下命令检查证书配置是否正确：

# 检查证书链完整性
openssl s_client -connect your-server.com:443 -showcerts

# 检查支持的TLS版本
nmap --script ssl-enum-ciphers -p 443 your-server.com

# 验证证书域名匹配
openssl x509 -in cert.pem -noout -text | grep -A1 "Subject Alternative Name"

3.3 性能优化配置

对于高并发的物联网应用，还需要对EMQX进行性能优化：

# 性能优化配置
node {
  process_limit = 2097152
  max_ports = 1048576
}

# 消息持久化配置（如果需要）
backend {
  redis {
    server = "127.0.0.1:6379"
    pool_size = 8
    database = 0
    password = ""
  }
}

# 连接管理优化
listeners.ssl.default {
  tcp_options {
    backlog = 1024
    send_timeout = "15s"
    send_timeout_close = true
    nodelay = true
    reuseaddr = true
  }
}

3.4 监控与告警配置

生产环境必须配置监控和告警，以下是一个基本的监控配置示例：

// EMQX监控配置
module.exports = {
  // 连接数监控
  connections: {
    warning_threshold: 1000,
    critical_threshold: 5000,
    check_interval: 5000  // 5秒检查一次
  },
  
  // 消息速率监控
  message_rate: {
    publish_warning: 100,    // 每秒发布消息警告阈值
    publish_critical: 500,   // 每秒发布消息紧急阈值
    deliver_warning: 1000,   // 每秒投递消息警告阈值
    deliver_critical: 5000   // 每秒投递消息紧急阈值
  },
  
  // 系统资源监控
  system: {
    cpu_warning: 80,         // CPU使用率警告阈值
    memory_warning: 85,      // 内存使用率警告阈值
    disk_warning: 90         // 磁盘使用率警告阈值
  }
}

4. 实战：完整的UniApp MQTT组件实现

现在让我们把这些知识整合起来，实现一个完整的、生产可用的UniApp MQTT组件。

4.1 组件架构设计

我设计了一个三层架构的MQTT组件：

1. 连接管理层：处理连接建立、重连、断开等基础功能
2. 消息路由层：管理主题订阅、消息分发、QoS处理
3. 业务适配层：提供业务友好的API接口

// mqtt-manager.js - 核心管理类
class MQTTManager {
  constructor(options = {}) {
    this.options = {
      host: '',
      port: 8083,
      path: '/mqtt',
      useSSL: true,
      username: '',
      password: '',
      clientId: '',
      clean: true,
      reconnectPeriod: 5000,
      connectTimeout: 30000,
      keepalive: 60,
      ...options
    }
    
    this.client = null
    this.subscriptions = new Map()  // 主题 -> 回调函数数组
    this.connectionState = 'disconnected'
    this.reconnectAttempts = 0
    this.maxReconnectAttempts = 10
    
    // 平台检测
    this.platform = this.detectPlatform()
  }
  
  // 平台检测方法
  detectPlatform() {
    let platform = 'unknown'
    
    // #ifdef H5
    platform = 'h5'
    // #endif
    
    // #ifdef MP-WEIXIN
    platform = 'mp-weixin'
    // #endif
    
    // #ifdef APP-PLUS
    platform = 'app-plus'
    // #endif
    
    return platform
  }
  
  // 构建连接URL
  buildConnectionUrl() {
    const { host, port, path, useSSL } = this.options
    
    switch(this.platform) {
      case 'h5':
        const protocol = useSSL ? 'wss' : 'ws'
        return `${protocol}://${host}:${port}${path}`
        
      case 'mp-weixin':
        const wxProtocol = useSSL ? 'wxs' : 'wx'
        // 微信小程序SSL连接必须使用443端口
        const wxPort = useSSL ? 443 : port
        return `${wxProtocol}://${host}:${wxPort}${path}`
        
      case 'app-plus':
        // APP端可以使用TCP或WebSocket
        const appProtocol = useSSL ? 'wss' : 'ws'
        return `${appProtocol}://${host}:${port}${path}`
        
      default:
        throw new Error(`不支持的平台: ${this.platform}`)
    }
  }
  
  // 连接方法
  async connect() {
    if (this.connectionState === 'connecting' || this.connectionState === 'connected') {
      console.warn('MQTT客户端已经在连接或已连接')
      return false
    }
    
    this.connectionState = 'connecting'
    
    try {
      const url = this.buildConnectionUrl()
      const options = this.buildConnectionOptions()
      
      // 动态导入mqtt库，避免打包问题
      const mqtt = await this.loadMQTTLibrary()
      
      this.client = mqtt.connect(url, options)
      this.setupEventHandlers()
      
      return new Promise((resolve, reject) => {
        const connectTimeout = setTimeout(() => {
          reject(new Error('MQTT连接超时'))
        }, this.options.connectTimeout)
        
        this.client.once('connect', () => {
          clearTimeout(connectTimeout)
          this.connectionState = 'connected'
          this.reconnectAttempts = 0
          console.log('MQTT连接成功')
          resolve(true)
        })
        
        this.client.once('error', (error) => {
          clearTimeout(connectTimeout)
          this.connectionState = 'error'
          reject(error)
        })
      })
    } catch (error) {
      this.connectionState = 'error'
      console.error('MQTT连接失败:', error)
      throw error
    }
  }
  
  // 动态加载MQTT库
  async loadMQTTLibrary() {
    // 根据平台选择不同的加载方式
    if (this.platform === 'mp-weixin') {
      // 微信小程序环境
      return require('./utils/mqtt.min.js')
    } else {
      // H5和APP环境
      return (await import('mqtt/dist/mqtt.js')).default
    }
  }
  
  // 构建连接选项
  buildConnectionOptions() {
    // 生成唯一的clientId，避免重复连接问题
    const clientId = this.options.clientId || 
      `mqtt_${this.platform}_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`
    
    return {
      clientId,
      username: this.options.username,
      password: this.options.password,
      clean: this.options.clean,
      reconnectPeriod: this.options.reconnectPeriod,
      connectTimeout: this.options.connectTimeout,
      keepalive: this.options.keepalive,
      resubscribe: true,
      properties: {
        sessionExpiryInterval: 3600, // 1小时会话过期
        receiveMaximum: 65535,       // 最大接收数量
        maximumPacketSize: 268435455 // 最大包大小
      }
    }
  }
  
  // 设置事件处理器
  setupEventHandlers() {
    if (!this.client) return
    
    this.client.on('connect', () => {
      console.log('MQTT已连接')
      this.connectionState = 'connected'
      this.reconnectAttempts = 0
      
      // 重新订阅之前订阅的主题
      this.resubscribeAll()
    })
    
    this.client.on('reconnect', () => {
      console.log('MQTT正在重连...')
      this.connectionState = 'reconnecting'
      this.reconnectAttempts++
      
      if (this.reconnectAttempts > this.maxReconnectAttempts) {
        console.error('MQTT重连次数超过限制，停止重连')
        this.client.end(true) // 强制断开
      }
    })
    
    this.client.on('close', () => {
      console.log('MQTT连接已关闭')
      this.connectionState = 'disconnected'
    })
    
    this.client.on('offline', () => {
      console.log('MQTT客户端离线')
      this.connectionState = 'offline'
    })
    
    this.client.on('error', (error) => {
      console.error('MQTT错误:', error)
      this.connectionState = 'error'
    })
    
    this.client.on('message', (topic, message) => {
      this.handleIncomingMessage(topic, message)
    })
  }
  
  // 处理接收到的消息
  handleIncomingMessage(topic, message) {
    const callbacks = this.subscriptions.get(topic) || []
    
    if (callbacks.length === 0) {
      // 如果没有订阅者，检查通配符订阅
      this.handleWildcardMessage(topic, message)
      return
    }
    
    // 调用所有注册的回调函数
    callbacks.forEach(callback => {
      try {
        callback(topic, message.toString())
      } catch (error) {
        console.error(`处理消息回调时出错 (主题: ${topic}):`, error)
      }
    })
  }
  
  // 处理通配符消息
  handleWildcardMessage(topic, message) {
    // 查找匹配的通配符订阅
    for (const [subTopic, callbacks] of this.subscriptions.entries()) {
      if (this.topicMatches(subTopic, topic)) {
        callbacks.forEach(callback => {
          try {
            callback(topic, message.toString())
          } catch (error) {
            console.error(`处理通配符消息回调时出错 (主题: ${subTopic}):`, error)
          }
        })
      }
    }
  }
  
  // 主题匹配检查（支持通配符）
  topicMatches(subscription, topic) {
    const subParts = subscription.split('/')
    const topicParts = topic.split('/')
    
    for (let i = 0; i < subParts.length; i++) {
      const subPart = subParts[i]
      const topicPart = topicParts[i]
      
      if (subPart === '#') {
        return true // 多级通配符
      }
      
      if (subPart === '+') {
        continue // 单级通配符
      }
      
      if (subPart !== topicPart) {
        return false
      }
    }
    
    return subParts.length === topicParts.length
  }
  
  // 订阅主题
  subscribe(topic, callback, options = {}) {
    if (!this.client || !this.client.connected) {
      console.warn('MQTT客户端未连接，无法订阅主题')
      return false
    }
    
    // 存储回调函数
    if (!this.subscriptions.has(topic)) {
      this.subscriptions.set(topic, [])
    }
    this.subscriptions.get(topic).push(callback)
    
    // 实际订阅
    const subscribeOptions = {
      qos: options.qos || 0,
      ...options
    }
    
    this.client.subscribe(topic, subscribeOptions, (error) => {
      if (error) {
        console.error(`订阅主题失败: ${topic}`, error)
        // 从订阅列表中移除失败的回调
        const callbacks = this.subscriptions.get(topic)
        if (callbacks) {
          const index = callbacks.indexOf(callback)
          if (index > -1) {
            callbacks.splice(index, 1)
          }
          if (callbacks.length === 0) {
            this.subscriptions.delete(topic)
          }
        }
      } else {
        console.log(`成功订阅主题: ${topic}`)
      }
    })
    
    return true
  }
  
  // 取消订阅
  unsubscribe(topic, callback = null) {
    if (!this.client || !this.client.connected) {
      console.warn('MQTT客户端未连接，无法取消订阅')
      return false
    }
    
    if (callback) {
      // 只移除特定的回调函数
      const callbacks = this.subscriptions.get(topic)
      if (callbacks) {
        const index = callbacks.indexOf(callback)
        if (index > -1) {
          callbacks.splice(index, 1)
        }
        if (callbacks.length === 0) {
          this.subscriptions.delete(topic)
          // 如果没有其他回调，实际取消订阅
          this.client.unsubscribe(topic)
        }
      }
    } else {
      // 移除所有回调并取消订阅
      this.subscriptions.delete(topic)
      this.client.unsubscribe(topic)
    }
    
    return true
  }
  
  // 发布消息
  publish(topic, message, options = {}) {
    if (!this.client || !this.client.connected) {
      console.warn('MQTT客户端未连接，无法发布消息')
      return false
    }
    
    const publishOptions = {
      qos: options.qos || 0,
      retain: options.retain || false,
      ...options
    }
    
    this.client.publish(topic, message, publishOptions, (error) => {
      if (error) {
        console.error(`发布消息失败 (主题: ${topic}):`, error)
      }
    })
    
    return true
  }
  
  // 重新订阅所有主题
  resubscribeAll() {
    for (const [topic, callbacks] of this.subscriptions.entries()) {
      if (callbacks.length > 0) {
        this.client.subscribe(topic, { qos: 0 }, (error) => {
          if (error) {
            console.error(`重新订阅失败: ${topic}`, error)
          } else {
            console.log(`重新订阅成功: ${topic}`)
          }
        })
      }
    }
  }
  
  // 断开连接
  disconnect() {
    if (this.client) {
      this.client.end(true) // true表示强制断开
      this.client = null
      this.connectionState = 'disconnected'
      this.subscriptions.clear()
      console.log('MQTT连接已断开')
    }
  }
  
  // 获取连接状态
  getConnectionState() {
    return this.connectionState
  }
  
  // 检查是否已连接
  isConnected() {
    return this.client && this.client.connected
  }
}

// 导出单例实例
let mqttInstance = null

export function getMQTTInstance(options) {
  if (!mqttInstance) {
    mqttInstance = new MQTTManager(options)
  }
  return mqttInstance
}

export default MQTTManager

4.2 Vue组件封装

为了让MQTT在Vue组件中更易用，我们可以创建一个Vue插件：

// mqtt-plugin.js
import { getMQTTInstance } from './mqtt-manager'

const MQTTPlugin = {
  install(Vue, options) {
    // 创建MQTT实例
    const mqtt = getMQTTInstance(options)
    
    // 添加到Vue原型
    Vue.prototype.$mqtt = mqtt
    
    // 添加全局混入，自动管理组件订阅
    Vue.mixin({
      created() {
        if (this.$options.mqttSubscriptions) {
          this._mqttSubscriptions = []
          
          Object.entries(this.$options.mqttSubscriptions).forEach(([topic, handler]) => {
            const callback = (topic, message) => {
              handler.call(this, message, topic)
            }
            
            if (mqtt.subscribe(topic, callback)) {
              this._mqttSubscriptions.push({ topic, callback })
            }
          })
        }
      },
      
      beforeDestroy() {
        if (this._mqttSubscriptions) {
          this._mqttSubscriptions.forEach(({ topic, callback }) => {
            mqtt.unsubscribe(topic, callback)
          })
          this._mqttSubscriptions = null
        }
      }
    })
    
    // 添加连接状态响应式属性
    Vue.prototype.$mqttState = Vue.observable({
      connected: false,
      connecting: false,
      error: null
    })
    
    // 监听连接状态变化
    mqtt.client.on('connect', () => {
      Vue.prototype.$mqttState.connected = true
      Vue.prototype.$mqttState.connecting = false
      Vue.prototype.$mqttState.error = null
    })
    
    mqtt.client.on('reconnect', () => {
      Vue.prototype.$mqttState.connected = false
      Vue.prototype.$mqttState.connecting = true
    })
    
    mqtt.client.on('error', (error) => {
      Vue.prototype.$mqttState.error = error
      Vue.prototype.$mqttState.connected = false
      Vue.prototype.$mqttState.connecting = false
    })
  }
}

export default MQTTPlugin

4.3 在Vue组件中使用

<template>
  <view class="mqtt-demo">
    <view class="status">
      连接状态: {{ connectionStatus }}
      <text v-if="error" class="error">错误: {{ error }}</text>
    </view>
    
    <view class="controls">
      <button @click="connect" :disabled="isConnecting || isConnected">
        {{ isConnecting ? '连接中...' : '连接' }}
      </button>
      
      <button @click="disconnect" :disabled="!isConnected">
        断开连接
      </button>
      
      <button @click="publishTest" :disabled="!isConnected">
        发布测试消息
      </button>
    </view>
    
    <view class="messages">
      <text class="section-title">接收到的消息:</text>
      <scroll-view scroll-y class="message-list">
        <view v-for="(msg, index) in messages" :key="index" class="message-item">
          <text class="topic">[{{ msg.topic }}]</text>
          <text class="content">{{ msg.content }}</text>
          <text class="time">{{ msg.time }}</text>
        </view>
      </scroll-view>
    </view>
    
    <view class="publish-form">
      <text class="section-title">发布消息</text>
      <input v-model="publishTopic" placeholder="输入主题" class="input" />
      <input v-model="publishMessage" placeholder="输入消息内容" class="input" />
      <button @click="publishCustom" :disabled="!isConnected">发布</button>
    </view>
  </view>
</template>

<script>
export default {
  data() {
    return {
      messages: [],
      publishTopic: 'test/topic',
      publishMessage: 'Hello MQTT!',
      error: null
    }
  },
  
  computed: {
    connectionStatus() {
      if (this.$mqttState.connected) return '已连接'
      if (this.$mqttState.connecting) return '连接中'
      return '未连接'
    },
    
    isConnected() {
      return this.$mqttState.connected
    },
    
    isConnecting() {
      return this.$mqttState.connecting
    }
  },
  
  // MQTT订阅配置
  mqttSubscriptions: {
    'test/#': function(message, topic) {
      this.onMessageReceived(topic, message)
    },
    
    'status/#': function(message, topic) {
      this.onStatusMessage(topic, message)
    }
  },
  
  methods: {
    async connect() {
      try {
        this.error = null
        await this.$mqtt.connect()
        console.log('MQTT连接成功')
      } catch (err) {
        this.error = err.message
        console.error('MQTT连接失败:', err)
      }
    },
    
    disconnect() {
      this.$mqtt.disconnect()
    },
    
    publishTest() {
      const topic = 'test/message'
      const message = JSON.stringify({
        type: 'test',
        timestamp: Date.now(),
        data: '这是一个测试消息'
      })
      
      this.$mqtt.publish(topic, message, { qos: 1 })
      this.addMessage(topic, message, 'sent')
    },
    
    publishCustom() {
      if (!this.publishTopic || !this.publishMessage) {
        uni.showToast({
          title: '请填写主题和消息',
          icon: 'none'
        })
        return
      }
      
      this.$mqtt.publish(this.publishTopic, this.publishMessage, { qos: 1 })
      this.addMessage(this.publishTopic, this.publishMessage, 'sent')
      
      // 清空输入
      this.publishMessage = ''
    },
    
    onMessageReceived(topic, message) {
      console.log(`收到消息 [${topic}]:`, message)
      this.addMessage(topic, message, 'received')
      
      // 显示通知
      uni.showToast({
        title: `收到新消息: ${topic}`,
        icon: 'none',
        duration: 2000
      })
    },
    
    onStatusMessage(topic, message) {
      console.log(`收到状态消息 [${topic}]:`, message)
      // 处理状态消息
      try {
        const data = JSON.parse(message)
        if (data.type === 'connection_count') {
          console.log(`当前连接数: ${data.count}`)
        }
      } catch (e) {
        console.warn('解析状态消息失败:', e)
      }
    },
    
    addMessage(topic, content, type) {
      this.messages.unshift({
        topic,
        content: typeof content === 'string' ? content : JSON.stringify(content),
        time: new Date().toLocaleTimeString(),
        type
      })
      
      // 限制消息列表长度
      if (this.messages.length > 50) {
        this.messages = this.messages.slice(0, 50)
      }
    }
  },
  
  mounted() {
    // 组件挂载时自动连接
    this.connect()
  },
  
  beforeDestroy() {
    // 组件销毁时断开连接
    this.disconnect()
  }
}
</script>

<style scoped>
.mqtt-demo {
  padding: 20rpx;
}

.status {
  padding: 20rpx;
  background-color: #f5f5f5;
  border-radius: 10rpx;
  margin-bottom: 20rpx;
}

.error {
  color: #ff4444;
  margin-left: 20rpx;
}

.controls {
  display: flex;
  flex-wrap: wrap;
  gap: 20rpx;
  margin-bottom: 30rpx;
}

.controls button {
  flex: 1;
  min-width: 150rpx;
}

.messages {
  margin-bottom: 30rpx;
}

.section-title {
  display: block;
  font-weight: bold;
  margin-bottom: 10rpx;
  color: #333;
}

.message-list {
  height: 300rpx;
  border: 1rpx solid #ddd;
  border-radius: 10rpx;
  padding: 10rpx;
  background-color: #fafafa;
}

.message-item {
  padding: 10rpx;
  border-bottom: 1rpx solid #eee;
}

.message-item:last-child {
  border-bottom: none;
}

.topic {
  color: #007aff;
  font-size: 24rpx;
  margin-right: 10rpx;
}

.content {
  font-size: 28rpx;
  color: #333;
}

.time {
  font-size: 22rpx;
  color: #999;
  float: right;
}

.publish-form {
  margin-top: 30rpx;
}

.input {
  border: 1rpx solid #ddd;
  border-radius: 8rpx;
  padding: 20rpx;
  margin-bottom: 20rpx;
  font-size: 28rpx;
}

.publish-form button {
  width: 100%;
}
</style>

4.4 全局状态管理集成

对于大型应用，建议将MQTT状态集成到Vuex或Pinia中：

// store/mqtt.js - Vuex模块
export default {
  namespaced: true,
  
  state: () => ({
    connected: false,
    connecting: false,
    error: null,
    messages: [],
    subscriptions: []
  }),
  
  mutations: {
    SET_CONNECTED(state, connected) {
      state.connected = connected
    },
    
    SET_CONNECTING(state, connecting) {
      state.connecting = connecting
    },
    
    SET_ERROR(state, error) {
      state.error = error
    },
    
    ADD_MESSAGE(state, message) {
      state.messages.unshift(message)
      // 限制消息数量
      if (state.messages.length > 100) {
        state.messages = state.messages.slice(0, 100)
      }
    },
    
    ADD_SUBSCRIPTION(state, subscription) {
      state.subscriptions.push(subscription)
    },
    
    REMOVE_SUBSCRIPTION(state, topic) {
      state.subscriptions = state.subscriptions.filter(sub => sub.topic !== topic)
    },
    
    CLEAR_MESSAGES(state) {
      state.messages = []
    }
  },
  
  actions: {
    async connect({ commit, rootState }) {
      commit('SET_CONNECTING', true)
      commit('SET_ERROR', null)
      
      try {
        const mqtt = getMQTTInstance(rootState.config.mqtt)
        await mqtt.connect()
        commit('SET_CONNECTED', true)
        commit('SET_CONNECTING', false)
        
        // 设置全局监听
        mqtt.client.on('message', (topic, message) => {
          commit('ADD_MESSAGE', {
            topic,
            content: message.toString(),
            timestamp: Date.now(),
            direction: 'incoming'
          })
        })
        
        return true
      } catch (error) {
        commit('SET_ERROR', error.message)
        commit('SET_CONNECTING', false)
        commit('SET_CONNECTED', false)
        return false
      }
    },
    
    disconnect({ commit, rootState }) {
      const mqtt = getMQTTInstance(rootState.config.mqtt)
      mqtt.disconnect()
      commit('SET_CONNECTED', false)
      commit('SET_CONNECTING', false)
    },
    
    subscribe({ commit, rootState }, { topic, callback }) {
      const mqtt = getMQTTInstance(rootState.config.mqtt)
      const success = mqtt.subscribe(topic, callback)
      
      if (success) {
        commit('ADD_SUBSCRIPTION', { topic, callback })
      }
      
      return success
    },
    
    unsubscribe({ commit, rootState }, topic) {
      const mqtt = getMQTTInstance(rootState.config.mqtt)
      const success = mqtt.unsubscribe(topic)
      
      if (success) {
        commit('REMOVE_SUBSCRIPTION', topic)
      }
      
      return success
    },
    
    publish({ rootState }, { topic, message, options }) {
      const mqtt = getMQTTInstance(rootState.config.mqtt)
      const success = mqtt.publish(topic, message, options)
      
      if (success) {
        // 记录发出的消息
        this.commit('mqtt/ADD_MESSAGE', {
          topic,
          content: typeof message === 'string' ? message : JSON.stringify(message),
          timestamp: Date.now(),
          direction: 'outgoing'
        })
      }
      
      return success
    }
  },
  
  getters: {
    isConnected: state => state.connected,
    isConnecting: state => state.connecting,
    lastError: state => state.error,
    recentMessages: state => state.messages.slice(0, 20),
    subscriptionCount: state => state.subscriptions.length
  }
}

5. 调试技巧与问题排查

即使有了完善的代码，在实际开发中还是会遇到各种问题。这里分享一些我在项目中积累的调试技巧。

5.1 真机调试连接失败排查

真机调试连接失败是最常见的问题，可以按照以下步骤排查：

1. 检查域名配置

   • 确保域名已在小程序后台配置
   • 域名必须经过ICP备案
   • SSL证书必须有效且完整

2. 网络环境检查

   // 网络状态检查工具
   class NetworkChecker {
     static async checkConnectivity() {
       return new Promise((resolve) => {
         // #ifdef MP-WEIXIN
         wx.getNetworkType({
           success: (res) => {
             console.log('当前网络类型:', res.networkType)
             resolve(res.networkType !== 'none')
           },
           fail: () => resolve(false)
         })
         // #endif
         
         // #ifdef H5
         resolve(navigator.onLine)
         // #endif
       })
     }
     
     static async checkServerReachable(url) {
       try {
         const response = await fetch(url, { method: 'HEAD' })
         return response.ok
       } catch (error) {
         console.error('服务器不可达:', error)
         return false
       }
     }
   }
   

3. 协议和端口检查

   • H5使用 ws:// 或 wss://
   • 小程序使用 wx:// 或 wxs://
   • SSL连接必须使用443端口

5.2 性能监控与优化

对于需要处理大量消息的应用，性能监控很重要：

// 性能监控工具
class MQTTPerformanceMonitor {
  constructor() {
    this.metrics = {
      messagesReceived: 0,
      messagesSent: 0,
      connectionTime: 0,
      lastMessageTime: 0,
      errors: []
    }
    
    this.startTime = Date.now()
  }
  
  recordMessageReceived() {
    this.metrics.messagesReceived++
    this.metrics.lastMessageTime = Date.now()
  }
  
  recordMessageSent() {
    this.metrics.messagesSent++
  }
  
  recordError(error) {
    this.metrics.errors.push({
      time: Date.now(),
      error: error.message || error.toString()
    })
    
    // 只保留最近100个错误
    if (this.metrics.errors.length > 100) {
      this.metrics.errors.shift()
    }
  }
  
  getStats() {
    const now = Date.now()
    const uptime = now - this.startTime
    
    return {
      ...this.metrics,
      uptime,
      messagesPerSecond: this.metrics.messagesReceived / (uptime / 1000),
      avgMessageInterval: this.metrics.messagesReceived > 0 
        ? uptime / this.metrics.messagesReceived 
        : 0
    }
  }
  
  // 生成性能报告
  generateReport() {
    const stats = this.getStats()
    const report = []
    
    report.push('=== MQTT性能报告 ===')
    report.push(`运行时间: ${Math.round(stats.uptime / 1000)}秒`)
    report.push(`接收消息: ${stats.messagesReceived}条`)
    report.push(`发送消息: ${stats.messagesSent}条`)
    report.push(`消息速率: ${stats.messagesPerSecond.toFixed(2)}条/秒`)
    report.push(`平均间隔: ${stats.avgMessageInterval.toFixed(0)}ms`)
    
    if (stats.errors.length > 0) {
      report.push(`\n最近错误 (${stats.errors.length}个):`)
      stats.errors.slice(-5).forEach((err, index) => {
        const time = new Date(err.time).toLocaleTimeString()
        report.push(`${index + 1}. [${time}] ${err.error}`)
      })
    }
    
    return report.join('\n')
  }
}

// 使用示例
const monitor = new MQTTPerformanceMonitor()

// 在消息处理中记录
mqttClient.on('message', (topic, message) => {
  monitor.recordMessageReceived()
  // ...处理消息
})

// 定期输出报告
setInterval(() => {
  console.log(monitor.generateReport())
}, 60000) // 每分钟输出一次

5.3 常见问题解决方案

根据我的经验，以下是一些常见问题及其解决方案：

问题1：小程序真机调试连接失败，但开发工具正常

原因：通常是SSL证书问题或域名未配置

解决方案：

1. 检查小程序后台域名配置
2. 验证SSL证书链完整性
3. 使用在线工具检查TLS配置
4. 确保使用正确的端口（SSL必须用443）

问题2：连接频繁断开重连

原因：网络不稳定或心跳配置不当

解决方案：

const options = {
  keepalive: 30,           // 更短的心跳间隔
  reconnectPeriod: 3000,   // 更快的重连间隔
  connectTimeout: 15000,   // 适当的超时时间
  clean: false             // 保持会话
}

问题3：消息丢失或重复

原因：QoS配置不当或客户端ID重复

解决方案：

1. 使用QoS 1或2确保消息可靠性
2. 生成唯一的客户端ID
3. 实现消息去重逻辑

// 消息去重实现
class MessageDeduplicator {
  constructor(windowSize = 1000, ttl = 60000) {
    this.receivedMessages = new Map()
    this.windowSize = windowSize
    this.ttl = ttl
  }
  
  isDuplicate(messageId) {
    // 清理过期记录
    this.cleanup()
    
    if (this.receivedMessages.has(messageId)) {
      return true
    }
    
    // 添加新记录
    this.receivedMessages.set(messageId, Date.now())
    
    // 限制记录数量
    if (this.receivedMessages.size > this.windowSize) {
      const oldestKey = this.receivedMessages.keys().next().value
      this.receivedMessages.delete(oldestKey)
    }
    
    return false
  }
  
  cleanup() {
    const now = Date.now()
    for (const [messageId, timestamp] of this.receivedMessages.entries()) {
      if (now - timestamp > this.ttl) {
        this.receivedMessages.delete(messageId)
      }
    }
  }
}

问题4：内存泄漏

原因：事件监听器未正确清理

解决方案：

class SafeMQTTClient {
  constructor() {
    this.eventHandlers = new Map()
  }
  
  on(event, handler) {
    if (!this.eventHandlers.has(event)) {
      this.eventHandlers.set(event, [])
    }
    this.eventHandlers.get(event).push(handler)

// 实际绑定到MQTT客户端 this.client.on(event, handler)

return () => this.off(event, handler)

}

off(event, handler) { const handlers = this.eventHandlers.get(event) if (handlers) { const index = handlers.indexOf(handler) if (index > -1) { handlers.splice(index, 1) this.client.off(event, handler) } } }

destroy() { // 清理所有事件监听器 for (const [event, handlers] of this.eventHandlers.entries()) { handlers.forEach(handler => { this.client.off(event, handler) }) } this.eventHandlers.clear() this.client.end() } }

### 5.4 高级调试技巧

对于复杂的问题，可能需要更深入的调试：

```javascript
// 增强的调试工具
class MQTTDebugger {
  constructor(client, options = {}) {
    this.client = client
    this.options = {
      logLevel: 'debug', // debug, info, warn, error
      logToConsole: true,
      logToStorage: false,
      maxLogEntries: 1000,
      ...options
    }
    
    this.logs = []
    this.setupLogging()
  }
  
  setupLogging() {
    const originalEmit = this.client.emit.bind(this.client)
    
    this.client.emit = (event, ...args) => {
      this.logEvent(event, args)
      return originalEmit(event, ...args)
    }
    
    // 监听所有事件
    const events = ['connect', 'reconnect', 'close', 'offline', 'error', 'message', 'packetsend', 'packetreceive']
    events.forEach(event => {
      this.client.on(event, (...args) => {
        this.logEvent(`client:${event}`, args)
      })
    })
  }
  
  logEvent(event, data) {
    const entry = {
      timestamp: Date.now(),
      event,
      data: this.sanitizeData(data),
      platform: this.getPlatform()
    }
    
    this.logs.push(entry)
    
    // 限制日志数量
    if (this.logs.length > this.options.maxLogEntries) {
      this.logs.shift()
    }
    
    // 输出到控制台
    if (this.options.logToConsole && this.shouldLog(event)) {
      console.log(`[MQTT Debug] ${new Date(entry.timestamp).toISOString()} ${event}:`, entry.data)
    }
    
    // 存储到本地
    if (this.options.logToStorage) {
      this.saveToStorage(entry)
    }
  }
  
  sanitizeData(data) {
    // 安全地序列化数据，避免循环引用
    try {
      return JSON.parse(JSON.stringify(data, (key, value) => {
        if (typeof value === 'function') return '[Function]'
        if (value instanceof Error) return { message: value.message, stack: value.stack }
        if (value instanceof Buffer) return `[Buffer ${value.length} bytes]`
        return value
      }))
    } catch (error) {
      return `[无法序列化: ${error.message}]`
    }
  }
  
  getPlatform() {
    // #ifdef H5
    return 'H5'
    // #endif
    // #ifdef MP-WEIXIN
    return 'MP-WEIXIN'
    // #endif
    // #ifdef APP-PLUS
    return 'APP-PLUS'
    // #endif
    return 'UNKNOWN'
  }
  
  shouldLog(event) {
    const levelMap = {
      debug: ['connect', 'reconnect', 'close', 'offline', 'packetsend', 'packetreceive'],
      info: ['message'],
      warn: ['reconnect'],
      error: ['error']
    }
    
    const currentLevel = this.options.logLevel
    const levels = ['debug', 'info', 'warn', 'error']
    const currentIndex = levels.indexOf(currentLevel)
    
    for (let i = currentIndex; i < levels.length; i++) {
      if (levelMap[levels[i]] && levelMap[levels[i]].includes(event)) {
        return true
      }
    }
    
    return false
  }
  
  saveToStorage(entry) {
    try {
      const storedLogs = uni.getStorageSync('mqtt_debug_logs') || []
      storedLogs.push(entry)
      
      // 限制存储数量
      if (storedLogs.length > this.options.maxLogEntries) {
        storedLogs.splice(0, storedLogs.length - this.options.maxLogEntries)
      }
      
      uni.setStorageSync('mqtt_debug_logs', storedLogs)
    } catch (error) {
      console.error('保存调试日志失败:', error)
    }
  }
  
  getLogs(filter = {}) {
    let filteredLogs = this.logs
    
    if (filter.startTime) {
      filteredLogs = filteredLogs.filter(log => log.timestamp >= filter.startTime)
    }
    
    if (filter.endTime) {
      filteredLogs = filteredLogs.filter(log => log.timestamp <= filter.endTime)
    }
    
    if (filter.event) {
      filteredLogs = filteredLogs.filter(log => log.event === filter.event)
    }
    
    if (filter.platform) {
      filteredLogs = filteredLogs.filter(log => log.platform === filter.platform)
    }
    
    return filteredLogs
  }
  
  exportLogs() {
    const logs = this.getLogs()
    const blob = new Blob([JSON.stringify(logs, null, 2)], { type: 'application/json' })
    const url = URL.createObjectURL(blob)
    
    // #ifdef H5
    const a = document.createElement('a')
    a.href = url
    a.download = `mqtt-debug-${Date.now()}.json`
    a.click()
    // #endif
    
    // #ifdef MP-WEIXIN || APP-PLUS
    // 在小程序和APP中，可能需要使用其他方式分享日志
    console.log('调试日志:', logs)
    // #endif
    
    URL.revokeObjectURL(url)
  }
  
  clearLogs() {
    this.logs = []
    if (this.options.logToStorage) {
      uni.removeStorageSync('mqtt_debug_logs')
    }
  }
}

// 使用示例
const mqttClient = getMQTTInstance(config)
const debugger = new MQTTDebugger(mqttClient.client, {
  logLevel: 'debug',
  logToConsole: true,
  logToStorage: true
})

// 在需要时导出日志
debugger.exportLogs()

5.5 网络质量检测与自适应

对于移动端应用，网络环境经常变化，需要实现自适应策略：

class NetworkAwareMQTTClient {
  constructor(config) {
    this.config = config
    this.client = null
    this.networkState = 'unknown'
    this.lastNetworkCheck = 0
    this.networkCheckInterval = 10000 // 10秒检查一次
    
    this.setupNetworkMonitoring()
  }
  
  setupNetworkMonitoring() {
    // #ifdef MP-WEIXIN
    wx.onNetworkStatusChange((res) => {
      this.handleNetworkChange(res.isConnected, res.networkType)
    })
    
    wx.getNetworkType({
      success: (res) => {
        this.handleNetworkChange(true, res.networkType)
      }
    })
    // #endif
    
    // #ifdef H5
    window.addEventListener('online', () => {
      this.handleNetworkChange(true, 'unknown')
    })
    
    window.addEventListener('offline', () => {
      this.handleNetworkChange(false, 'none')
    })
    
    this.handleNetworkChange(navigator.onLine, 'unknown')
    // #endif
    
    // 定期检查网络质量
    setInterval(() => {
      this.checkNetworkQuality()
    }, this.networkCheckInterval)
  }
  
  handleNetworkChange(isConnected, networkType) {
    const oldState = this.networkState
    this.networkState = isConnected ? networkType : 'disconnected'
    
    console.log(`网络状态变化: ${oldState} -> ${this.networkState}`)
    
    if (oldState === 'disconnected' && isConnected) {
      // 网络恢复，尝试重连
      this.reconnectIfNeeded()
    } else if (!isConnected && this.client && this.client.connected) {
      // 网络断开，记录状态但不立即断开
      console.log('网络断开，保持MQTT连接等待恢复')
    }
  }
  
  async checkNetworkQuality() {
    const now = Date.now()
    if (now - this.lastNetworkCheck < 5000) {
      return // 避免频繁检查
    }
    
    this.lastNetworkCheck = now
    
    try {
      const latency = await this.measureLatency()
      const throughput = await this.measureThroughput()
      
      this.adjustMQTTSettings(latency, throughput)
    } catch (error) {
      console.warn('网络质量检测失败:', error)
    }
  }
  
  async measureLatency() {
    const start = Date.now()
    
    // 简单的延迟测量
    return new Promise((resolve) => {
      // 这里可以实现具体的延迟测量逻辑
      // 例如发送一个小的ping包到服务器
      setTimeout(() => {
        resolve(Date.now() - start)
      }, 100)
    })
  }
  
  async measureThroughput() {
    // 简单的吞吐量测量
    return Promise.resolve(1024) // 默认值
  }
  
  adjustMQTTSettings(latency, throughput) {
    if (!this.client) return
    
    // 根据网络质量调整MQTT设置
    if (latency > 1000) {
      // 高延迟网络
      this.client.options.keepalive = 120 // 延长心跳间隔
      this.client.options.reconnectPeriod = 10000 // 延长重连间隔
      console.log('检测到高延迟网络，调整MQTT参数')
    } else if (throughput < 51200) {
      // 低带宽网络
      console.log('检测到低带宽网络，考虑压缩消息')
    } else {
      // 良好网络
      this.client.options.keepalive = 60
      this.client.options.reconnectPeriod = 5000
    }
  }
  
  reconnectIfNeeded() {
    if (this.client && !this.client.connected && this.networkState !== 'disconnected') {
      console.log('网络恢复，尝试重新连接MQTT')
      setTimeout(() => {
        if (this.client && !this.client.connected) {
          this.client.reconnect()
        }
      }, 1000) // 延迟1秒重连，避免网络不稳定
    }
  }
  
  // 包装原有的connect方法
  async connect() {
    if (this.networkState === 'disconnected') {
      throw new Error('网络未连接，无法建立MQTT连接')
    }
    
    // 原有的连接逻辑
    return super.connect()
  }
}

6. 生产环境最佳实践

在将MQTT集成到生产环境时，还需要考虑以下最佳实践：

6.1 安全配置

// 安全配置示例
const securityConfig = {
  // 使用Token认证
  authentication: {
    type: 'jwt', // 或 'token', 'certificate'
    token: async () => {
      // 从安全存储获取token
      const token = await secureStorage.get('mqtt_token')
      if (!token || this.isTokenExpired(token)) {
        return await this.refreshToken()
      }
      return token
    },
    refreshInterval: 3600000 // 1小时刷新一次
  },
  
  // 消息加密
  encryption: {
    enabled: true,
    algorithm: 'aes-256-gcm',
    keyRotationInterval: 86400000 // 24小时轮换一次
  },
  
  // 访问控制
  acl: {
    enabled: true,
    rules: [
      { topic: 'public/#', access: 'read' },
      { topic: 'private/${clientId}/#', access: 'readwrite' },
      { topic: 'admin/#', access: 'deny' }
    ]
  },
  
  // 审计日志
  audit: {
    enabled: true,
    logConnections: true,
    logDisconnections: true,
    logMessages: false, // 生产环境通常不记录消息内容
    logErrors: true
  }
}

6.2 错误处理与恢复

class ResilientMQTTClient {
  constructor(config) {
    this.config = config
    this.retryStrategy = this.createRetryStrategy()
    this.circuitBreaker = this.createCircuitBreaker()
    this.messageQueue = new MessageQueue()
  }
  
  createRetryStrategy() {
    return {
      maxRetries: 5,
      initialDelay: 1000,
      maxDelay: 30000,
      backoffFactor: 2,
      
      shouldRetry(error) {
        // 可重试的错误类型
        const retryableErrors = [
          'ECONNREFUSED',
          'ETIMEDOUT',
          'ENETUNREACH',
          'ECONNRESET'
        ]
        
        return retryableErrors.some(code => 
          error.code === code || error.message.includes(code)
        )
      },
      
      getDelay(retryCount) {
        const delay = this.initialDelay * Math.pow(this.backoffFactor, retryCount)
        return Math.min(delay, this.maxDelay)
      }
    }
  }
  
  createCircuitBreaker() {
    return {
      state: 'CLOSED', // CLOSED, OPEN, HALF_OPEN
      failureCount: 0,
      successCount: 0,
      threshold: 5,
      resetTimeout: 60000,
      lastFailureTime: 0,
      
      canExecute() {
        if (this.state === 'OPEN') {
          const now = Date.now()
          if (now - this.lastFailureTime > this.resetTimeout) {
            this.state = 'HALF_OPEN'
            return true
          }
          return false
        }
        return true
      },
      
      onSuccess() {
        if (this.state === 'HALF_OPEN') {
          this.successCount++
          if (this.successCount >= this.threshold) {
            this.state = 'CLOSED'
            this.failureCount = 0
            this.successCount = 0
          }
        }
      },
      
      onFailure() {
        this.failureCount++
        this.lastFailureTime = Date.now()
        
        if (this.state === 'CLOSED' && this.failureCount >= this.threshold) {
          this.state = 'OPEN'
          console.warn('断路器打开，暂时停止MQTT操作')
        } else if (this.state === 'HALF_OPEN') {
          this.state = 'OPEN'
        }
      }
    }
  }
  
  async connectWithRetry() {
    for (let retryCount = 0; retryCount <= this.retryStrategy.maxRetries; retryCount++) {
      if (!this.circuitBreaker.canExecute()) {
        throw new Error('断路器打开，无法连接')
      }
      
      try {
        const result = await this.connect()
        this.circuitBreaker.onSuccess()
        return result
      } catch (error) {
        this.circuitBreaker.onFailure()
        
        if (retryCount === this.retryStrategy.maxRetries || 
            !this.retryStrategy.shouldRetry(error)) {
          throw error
        }
        
        const delay = this.retryStrategy.getDelay(retryCount)
        console.log(`连接失败，${delay}ms后重试 (${retryCount + 1}/${this.retryStrategy.maxRetries})`)
        
        await this.sleep(delay)
      }
    }
  }
  
  async publishWithQueue(topic, message, options = {}) {
    if (!this.isConnected()) {
      // 将消息加入队列，等待连接恢复
      this.messageQueue.enqueue({ topic, message, options })
      return false
    }
    
    try {
      const result = await this.publish(topic, message, options)
      
      // 发送成功后，尝试发送队列中的消息
      this.processQueue()
      
      return result
    } catch (error) {
      // 发送失败，将消息重新加入队列
      this.messageQueue.enqueue({ topic, message, options })
      throw error
    }
  }
  
  async processQueue() {
    while (this.isConnected() && !this.messageQueue.isEmpty()) {
      const item = this.messageQueue.peek()
      
      try {
        await this.publish(item.topic, item.message, item.options)
        this.messageQueue.dequeue()
      } catch (error) {
        console.error('发送队列消息失败:', error)
        break
      }
    }
  }
  
  sleep(ms) {
    return new Promise(resolve => setTimeout(resolve, ms))
  }
}

6.3 性能监控与告警

class MQTTMonitor {
  constructor() {
    this.metrics = {
      connections: 0,
      messages: { in: 0, out: 0 },
      errors: [],
      latency: [],
      throughput: 0
    }
    
    this.alerts = new Map()
    this.setupMonitoring()
  }
  
  setupMonitoring() {
    // 定期收集指标
    setInterval(() => this.collectMetrics(), 60000)
    
    // 定期检查告警
    setInterval(() => this.checkAlerts(), 30000)
  }
  
  collectMetrics() {
    const now = Date.now()
    
    // 计算消息速率
    const duration = 60000 // 1分钟
    const messageRate = this.metrics.messages.in / (duration / 1000)
    this.metrics.throughput = messageRate
    
    // 保存历史数据
    this.saveSnapshot(now)
    
    // 重置计数器
    this.metrics.messages = { in: 0, out: 0 }
  }
  
  saveSnapshot(timestamp) {
    const snapshot = {
      timestamp,
      connections: this.metrics.connections,
      messageRate: this.metrics.throughput,
      avgLatency: this.calculateAverageLatency(),
      errorCount: this.metrics.errors.length
    }
    
    // 存储到本地或发送到监控服务器
    this.storeSnapshot(snapshot)
  }
  
  calculateAverageLatency() {
    if (this.metrics.latency.length === 0) return 0
    
    const sum = this.metrics.latency.reduce((a, b) => a + b, 0)
    return sum / this.metrics.latency.length
  }
  
  storeSnapshot(snapshot) {
    // 这里可以实现存储逻辑，如IndexedDB、LocalStorage或发送到服务器
    console.log('监控快照:', snapshot)
  }
  
  addAlert(name, condition, action) {
    this.alerts.set(name, { condition, action, triggered: false })
  }
  
  checkAlerts() {
    for (const [name, alert] of this.alerts.entries()) {
      const shouldTrigger = alert.condition(this.metrics)
      
      if (shouldTrigger && !alert.triggered) {
        alert.action(this.metrics)
        alert.triggered = true
      } else if (!shouldTrigger && alert.triggered) {
        alert.triggered = false
      }
    }
  }
  
  // 示例告警条件
  static createAlerts(monitor) {
    // 高消息速率告警
    monitor.addAlert(
      'high_message_rate',
      (metrics) => metrics.throughput > 1000, // 每秒超过1000条消息
      (metrics) => {
        console.warn(`高消息速率告警: ${metrics.throughput.toFixed(2)} 条/秒`)
        // 这里可以发送邮件、短信或调用其他告警接口
      }
    )
    
    // 高延迟告警
    monitor.addAlert(
      'high_latency',
      (metrics) => {
        const avgLatency = monitor.calculateAverageLatency()
        return avgLatency > 1000 // 平均延迟超过1秒
      },
      (metrics) => {
        const avgLatency = monitor.calculateAverageLatency()
        console.warn(`高延迟告警: ${avgLatency.toFixed(0)}ms`)
      }
    )
    
    // 错误频率告警
    monitor.addAlert(
      'high_error_rate',
      (metrics) => metrics.errors.length > 10, // 最近1分钟超过10个错误
      (metrics) => {
        console.warn(`高错误频率告警: ${metrics.errors.length} 个错误`)
      }
    )
  }
}

// 使用示例
const monitor = new MQTTMonitor()
MQTTMonitor.createAlerts(monitor)

// 在MQTT事件中更新指标
mqttClient.on('message', () => {
  monitor.metrics.messages.in++
})

mqttClient.on('publish', () => {
  monitor.metrics.messages.out++
})

mqttClient.on('error', (error) => {
  monitor.metrics.errors.push({
    timestamp: Date.now(),
    error: error.message
  })
})

6.4 部署与运维建议

1. 服务器部署

   • 使用Docker部署EMQX，便于管理和扩展
   • 配置合理的资源限制和健康检查
   • 设置日志轮转和监控告警

2. 客户端配置

   • 根据网络环境动态调整参数
   • 实现优雅降级和故障转移
   • 添加使用统计和错误报告

3. 监控与维护

   • 监控连接数、消息速率、错误率等关键指标
   • 定期检查证书有效期
   • 建立应急预案和回滚机制

4. 安全加固

   • 使用TLS加密所有通信
   • 实现客户端认证和授权
   • 定期更新依赖库和安全补丁

通过以上完整的实现和最佳实践，你应该能够构建出稳定、可靠、高效的UniApp MQTT应用。记住，实际项目中还需要根据具体需求进行调整和优化，但这里提供的框架和模式应该能为你节省大量时间和精力。