iOS应用订阅AWS IoT Core超50个Thing时设备随机离线的问题求助

我最近在开发一款iOS智能家居App，使用AWS IoT Core和MQTT协议管理设备，每个物理设备对应一个AWS IoT的Thing，App需要订阅每个Thing的影子更新delta主题：$aws/things/<thingName>/shadow/update/delta。

遇到了一个非常棘手的问题：这个问题仅在iOS App上出现，相同逻辑在Android App上连接80+设备都完全正常。当需要管理的设备数量超过50个时，App里会随机有6~13个设备显示为离线状态，但实际上这些设备都是在线的，只是它们的delta主题消息从来没有被App接收到。不过如果我关闭并重新打开App 2~3次，所有设备又都能正常显示在线，并且可以正确接收delta消息了。

我的实现细节 #

我会在订阅前先确保MQTT处于连接状态，核心的连接和订阅代码如下：

MQTT连接核心代码

func createIOTConnection() {
    if timer.isValid {
        timer.invalidate()
    }
    if InternetConnectionManager.shared.checkInternetConnection() {
        print("iot* createIOTConnection")
        count = 0
        let uuid = UUID().uuidString
        print("iot* uuid \(uuid)")
        debugPrint("ITC: MQTT Client UUID: \(uuid)")
        iotDataManager.connectUsingWebSocket(withClientId: uuid, cleanSession: true) { (connectionStatus) in
            switch connectionStatus {
            case .unknown:
                print("iot* unknown")
                self.showBannerWithMsg(msg: "Connecting")
            case .connecting:
                print("iot* connecting")
                self.showBannerWithMsg(msg: "Connecting")
            case .connected:
                print("iot* connected")
                self.showBannerWithMsg(msg: "Connected")
                self.subscribeForDelta()
                self.subscribeToAllModules(true)
                self.delegate?.connectedToIoT?()
                NotificationCenter.default.post(name: NSNotification.Name(rawValue: "iotConnected"), object: nil, userInfo: nil)
            case .disconnected:
                print("iot* disconnected")
                self.showBannerWithMsg(msg: "Connecting")
            case .connectionRefused:
                print("iot* connectionrefused")
            case .connectionError:
                print("iot* connectionerror")
                self.showBannerWithMsg(msg: "Connecting")
            case .protocolError:
                print("iot* protocolerror")
            }
        }
        timer = Timer.scheduledTimer(timeInterval: 5, target: self, selector: #selector(continuousCheckForIOTConnection), userInfo: nil, repeats: true)
    } else {
        timer = Timer.scheduledTimer(timeInterval: 1, target: self, selector: #selector(checkIfInternetIsAvailable), userInfo: nil, repeats: true)
    }
}

批量订阅设备主题代码

func subscribeToAllModules(_ getStatus:Bool) {
    if UserDefaults.standard.bool(forKey: "isLoggedIn") {
        DispatchQueue.global().async {
            let array = dbManager.fetchList(query: "select serialNumber from master", columnName: "serialNumber", .master) as! [String]
            for srnum in array {
                self.subscribeWithSerialnumber(serialnumber: srnum, operation: .update)
                if srnum.contains("zdp") {
                    self.subscribeWithSerialnumber(serialnumber: srnum, operation: .get)
                }
            }
            
            if getStatus && !isSceneExecuting {
                let modules = dbManager.fetchData("select serialNumber,moduleType from master", .master)
                modules.forEach { (module) in
                    let dict = Module_Data.getActiveModuleGroupsWithApplianceCount(moduleString: module["moduleType"]! as! String)
                    if dict.keys.contains("LI") || dict.keys.contains("DL") || dict.keys.contains("FA") || dict.keys.contains("PM") || dict.keys.contains("CM") {
                        let srnum = module["serialNumber"]! as! String
                        if (RoomDetails.roomExecution != nil && !RoomDetails.roomExecution.sentSerialNumbers.contains(srnum)) || RoomDetails.roomExecution == nil {
                            self.updateThing(thingname: srnum, json: self.pingMsg, shadowOperation: .update)
                        }
                    }
                }
            }
        }
    }
}

订阅用户专属Delta主题代码

func subscribeForDelta() {
    if let email = UserDefaults.standard.string(forKey: "email") {
        self.subscribeWithSerialnumber(serialnumber: email.alphanumeric, operation: .get)
        self.updateThing(thingname: email.alphanumeric, json: "", shadowOperation: .get)
        let topic = "$aws/things/\(email.alphanumeric)/shadow/update/delta"
        debugPrint("ITC: Subscribing for delta topic: \(topic)")
        let subscribed = iotDataManager!.subscribe(toTopic: topic, qoS: .messageDeliveryAttemptedAtMostOnce) { (data) in
            debugPrint("ITC: Delta payload received")
            debugPrint(data)
            do {
                let d = try JSONSerialization.jsonObject(with: data, options: [])
                // 原代码此处截断，保留现有处理逻辑框架
            } catch {
                print("解析delta消息失败: \(error.localizedDescription)")
            }
        }
    }
}

已做的排查动作 #

1. 每次订阅前都会严格确认MQTT处于connected状态
2. 每次连接使用随机生成的UUID作为clientId，且设置cleanSession: true
3. 批量订阅放在全局队列异步执行，避免阻塞主线程

现在我有点摸不着头绪，有没有大佬遇到过类似的问题？会不会是iOS端AWS IoT SDK有订阅数量的隐性限制？或者是批量订阅时的异步处理存在潜在问题，导致部分订阅请求失败？

内容来源于stack exchange