戰(zhàn))
去年接了個(gè)挺頭疼的活把公司一套基于 Flutter 的客戶端應(yīng)用遷移到 OpenHarmony 設(shè)備上。界面、狀態(tài)管理、本地存儲(chǔ)都順利解決了最后卡在一個(gè)叫aws_sqs_api的三方庫(kù)上。這個(gè)庫(kù)是 AWS SQSSimple Queue Service的 Dart 客戶端我們?cè)谠?Android/iOS 版本里用它做設(shè)備端數(shù)據(jù)上報(bào)把采集到的狀態(tài)、日志、業(yè)務(wù)事件一股腦丟進(jìn)云端隊(duì)列后端服務(wù)異步消費(fèi)實(shí)現(xiàn)分布式場(chǎng)景下的消息解耦和削峰填谷。搬到 OpenHarmony 后這個(gè)鏈路必須原樣跑通否則所有設(shè)備上報(bào)都會(huì)變成直連后端 HTTP 接口一旦流量抖動(dòng)后端就會(huì)被打爆。這篇文章就是那次完整適配過(guò)程的復(fù)盤里面包含方案取舍、MethodChannel 橋接細(xì)節(jié)、SigV4 簽名在 ArkTS 側(cè)的實(shí)現(xiàn)以及我在生產(chǎn)環(huán)境里踩過(guò)的坑希望能給同樣在 OpenHarmony 上做 Flutter 插件適配的同學(xué)省點(diǎn)時(shí)間。1. 這個(gè)庫(kù)到底是干什么的分布式消息異步解耦的切入點(diǎn)1.1 為什么選 SQS 而不是其他消息中間件先聊聊背景。我們的場(chǎng)景是物聯(lián)網(wǎng)設(shè)備端上報(bào)設(shè)備數(shù)量上千臺(tái)每臺(tái)每隔幾秒就會(huì)產(chǎn)生一條狀態(tài)數(shù)據(jù)。如果設(shè)備直連后端 API高峰期每秒可能有上千個(gè)并發(fā)請(qǐng)求后端服務(wù)要么瘋狂擴(kuò)容要么直接限流丟數(shù)據(jù)。用消息隊(duì)列做中轉(zhuǎn)以后設(shè)備端只負(fù)責(zé)把消息丟進(jìn)隊(duì)列后端按自己的處理能力去拉取兩端互不阻塞這就是典型的異步解耦。技術(shù)選型時(shí)我們對(duì)比過(guò) RabbitMQ、Kafka 和 AWS SQS。自建 RabbitMQ 或 Kafka 在云端要考慮運(yùn)維成本而且我們的客戶端是 Flutter 寫的需要找 Dart 生態(tài)里維護(hù)活躍的 SDK。AWS SQS 雖然是云廠商托管服務(wù)但勝在完全不用運(yùn)維標(biāo)準(zhǔn)隊(duì)列無(wú)限吞吐還有死信隊(duì)列、延遲隊(duì)列、長(zhǎng)輪詢這些開(kāi)箱即用的能力。配合aws_sqs_api這個(gè)純 Dart 包Dart 層直接調(diào)用 SQS 的 REST API省掉了中間再套一層自建網(wǎng)關(guān)的成本。1.2 aws_sqs_api 的功能邊界與依賴關(guān)系aws_sqs_api是 AWS 官方為 Dart 語(yǔ)言生成的 SQS API 客戶端底層用的是 AWS 的 Smithy 代碼生成框架。它本身不包含 UI 組件也不依賴任何 Flutter 原生插件核心能力就是封裝 SQS 的 REST API 調(diào)用包括創(chuàng)建隊(duì)列、發(fā)送消息、接收消息、刪除消息、修改可見(jiàn)性超時(shí)等。它有幾個(gè)關(guān)鍵依賴包需要一起引入aws_common提供 AWS 服務(wù)的通用基礎(chǔ)類型和配置aws_signature_v4實(shí)現(xiàn) AWS Signature Version 4 請(qǐng)求簽名aws_smithy_clientSmithy 客戶端運(yùn)行時(shí)負(fù)責(zé) HTTP 請(qǐng)求的發(fā)送和響應(yīng)解析這個(gè)依賴關(guān)系很重要。aws_signature_v4是純 Dart 實(shí)現(xiàn)的簽名算法理論上在任何能跑 Dart 的平臺(tái)上都能運(yùn)行。但實(shí)際適配 OpenHarmony 時(shí)問(wèn)題往往出在更底層——Dart 運(yùn)行時(shí)能不能正常發(fā) HTTPS 請(qǐng)求、網(wǎng)絡(luò)權(quán)限怎么配、憑證存哪里。搞清楚這些邊界你就知道鴻蒙適配的重點(diǎn)其實(shí)不在 Dart 層而在平臺(tái)橋接層。2. 鴻蒙適配的核心難點(diǎn)拆解2.1 三層問(wèn)題運(yùn)行時(shí)、簽名、原生通道把a(bǔ)ws_sqs_api搬到 OpenHarmony我把它拆成了三個(gè)層次的問(wèn)題逐個(gè)擊破第一層是 Dart 運(yùn)行時(shí)兼容性。OpenHarmony 上的 Flutter 是基于 OpenHarmony 官方移植的 Flutter SDK 來(lái)跑的大部分dart:io的能力都支持但跟 Android/iOS 的 Flutter 運(yùn)行時(shí)不是同一個(gè)實(shí)現(xiàn)。我們?cè)谶m配過(guò)程中發(fā)現(xiàn)aws_smithy_client里的某些網(wǎng)絡(luò)異常處理在 OpenHarmony 上的表現(xiàn)略有差異具體來(lái)說(shuō)是SocketException的報(bào)錯(cuò)信息格式不一樣導(dǎo)致日志解析邏輯需要微調(diào)。這一層的問(wèn)題比較隱蔽建議適配時(shí)先寫一個(gè)最小化的 Dart 腳本在目標(biāo)設(shè)備上跑一遍確認(rèn) HttpClient 能正常訪問(wèn)外網(wǎng)。第二層是 SigV4 簽名算法的平臺(tái)差異。aws_signature_v4用的是純 Dart 的crypto包做 SHA256 和 HMAC 計(jì)算在 OpenHarmony 的 Dart 運(yùn)行時(shí)上可以正常工作。但這里有個(gè)坑SQS 的請(qǐng)求簽名要求 CanonicalRequest 里的 host 頭必須和實(shí)際請(qǐng)求的 host 完全一致包括大小寫和端口號(hào)。在 OpenHarmony 上如果你走了代理或者自定義了網(wǎng)絡(luò)棧host 頭可能會(huì)被改寫導(dǎo)致服務(wù)端返回SignatureDoesNotMatch。第三層是原生平臺(tái)通道。項(xiàng)目里的憑證信息之前是存在系統(tǒng)鑰匙串里的Android 用的是flutter_secure_storageiOS 用 Keychain。OpenHarmony 上沒(méi)有現(xiàn)成的插件這層必須自己寫原生橋接。另外我們的業(yè)務(wù)還要求 App 在后臺(tái)時(shí)也能持續(xù)消費(fèi)隊(duì)列消息這需要鴻蒙端的任務(wù)后臺(tái)執(zhí)行能力配合不是一個(gè)純 Dart 包能解決的。所以適配工作的重心最終落在了 MethodChannel 的建聯(lián)和 ArkTS 原生側(cè)的實(shí)現(xiàn)上。2.2 MethodChannel 橋接 vs 純 Dart 直連的取舍有一種思路是既然aws_sqs_api是純 Dart 包OpenHarmony 的 Flutter 運(yùn)行時(shí)又支持dart:io那是不是什么都不用改直接跑就完事了我最初也是這么想的在開(kāi)發(fā)機(jī)上跑了個(gè) demo還真能通。但放到生產(chǎn)環(huán)境就暴露了三個(gè)問(wèn)題憑證存儲(chǔ)沒(méi)有安全的地方。Dart 側(cè)只能用 shared_preferences 之類的插件存明文這在合規(guī)審計(jì)上過(guò)不去。后臺(tái)消費(fèi)不可靠。Flutter 的 Dart isolate 在應(yīng)用退到后臺(tái)后可能被系統(tǒng)掛起沒(méi)有鴻蒙端原生任務(wù)配合消息消費(fèi)會(huì)斷。網(wǎng)絡(luò)棧不可控。某些定制 ROM 的 OpenHarmony 設(shè)備會(huì)對(duì) Flutter 的 HttpClient 做限制而走系統(tǒng)ohos.net.http是經(jīng)過(guò)充分驗(yàn)證的通道。所以最終方案是Dart 層通過(guò) MethodChannel 調(diào) ArkTS 原生實(shí)現(xiàn)把發(fā)送消息、接收消息、刪除消息、修改可見(jiàn)性這幾個(gè)核心操作全部下沉到鴻蒙側(cè)。aws_sqs_api在 Dart 層保留作為接口定義和數(shù)據(jù)模型參考真正發(fā) HTTP 請(qǐng)求的是 ArkTS 代碼。這個(gè)方案雖然多寫了不少原生代碼但換來(lái)的是安全存儲(chǔ)、穩(wěn)定網(wǎng)絡(luò)和后臺(tái)執(zhí)行能力我認(rèn)為是值得的。3. 實(shí)操?gòu)膭?chuàng)建插件工程到跑通第一條消息3.1 工程結(jié)構(gòu)配置與權(quán)限聲明OpenHarmony 的 Flutter 插件和 Android 插件結(jié)構(gòu)很像但目錄名從android換成了ohos。我用的是手動(dòng)創(chuàng)建的方式因?yàn)閒lutter create --templateplugin默認(rèn)不支持生成 ohos 目錄。工程目錄結(jié)構(gòu)長(zhǎng)這樣aws_sqs_api_ohos/ ├── pubspec.yaml ├── lib/ │ ├── aws_sqs_api_ohos.dart │ └── src/ │ └── (dart層方法通道封裝) └── ohos/ ├── build-profile.json5 └── entry/ └── src/ └── main/ ├── ets/ │ ├── entryability/ │ └── plugins/ │ └── AwsSqsApiPlugin.ets └── module.json5pubspec.yaml里要聲明插件支持的平臺(tái)注意要加上 ohosflutter: plugin: platforms: android: package: com.example.aws_sqs_api_ohos pluginClass: AwsSqsApiPlugin ios: pluginClass: AwsSqsApiPlugin ohos: pluginClass: AwsSqsApiPlugin pluginImplementation: AwsSqsApiPluginImplmodule.json5里必須聲明網(wǎng)絡(luò)權(quán)限這是最容易漏的一步。鴻蒙應(yīng)用默認(rèn)沒(méi)有網(wǎng)絡(luò)訪問(wèn)權(quán)限不加這個(gè)權(quán)限所有 HTTPS 請(qǐng)求都會(huì)靜默失敗{ module: { name: entry, requestPermissions: [ { name: ohos.permission.INTERNET } ] } }這個(gè)權(quán)限配置和 Android 的AndroidManifest.xml里加uses-permission android:nameandroid.permission.INTERNET /是同一個(gè)作用但位置完全不同很多從 Android 轉(zhuǎn)過(guò)來(lái)的同學(xué)會(huì)下意識(shí)去找 manifest 文件結(jié)果在鴻蒙工程里根本找不到。3.2 Dart 側(cè)封裝MethodChannel 的調(diào)用契約Dart 側(cè)的封裝盡量保持和原來(lái)aws_sqs_api的調(diào)用風(fēng)格一致這樣業(yè)務(wù)代碼不用大面積改動(dòng)。我定義了一個(gè)統(tǒng)一的方法通道名aws_sqs_api然后按 SQS 的核心操作拆成幾個(gè)方法。class AwsSqsApiOhos { static const MethodChannel _channel MethodChannel(aws_sqs_api); static FutureString sendMessage({ required String queueUrl, required String messageBody, int delaySeconds 0, MapString, String attributes const {}, }) async { final result await _channel.invokeMethod(sendMessage, { queueUrl: queueUrl, messageBody: messageBody, delaySeconds: delaySeconds, messageAttributes: attributes, }); return result as String; } static FutureListMapString, dynamic receiveMessage({ required String queueUrl, int maxNumberOfMessages 10, int waitTimeSeconds 0, int visibilityTimeout 30, }) async { final result await _channel.invokeMethod(receiveMessage, { queueUrl: queueUrl, maxNumberOfMessages: maxNumberOfMessages, waitTimeSeconds: waitTimeSeconds, visibilityTimeout: visibilityTimeout, }); return (result as List).castMapString, dynamic(); } static Futurebool deleteMessage({ required String queueUrl, required String receiptHandle, }) async { final result await _channel.invokeMethod(deleteMessage, { queueUrl: queueUrl, receiptHandle: receiptHandle, }); return result as bool; } static Futurevoid changeMessageVisibility({ required String queueUrl, required String receiptHandle, required int visibilityTimeout, }) async { await _channel.invokeMethod(changeMessageVisibility, { queueUrl: queueUrl, receiptHandle: receiptHandle, visibilityTimeout: visibilityTimeout, }); } }注意幾個(gè)設(shè)計(jì)細(xì)節(jié)receiveMessage的返回值我用了ListMapString, dynamic而不是強(qiáng)類型對(duì)象因?yàn)?MethodChannel 的 JSON 反序列化在鴻蒙端的Map鍵值類型可能和 Dart 側(cè)不完全匹配留一層動(dòng)態(tài)類型可以減少類型轉(zhuǎn)換異常。所有方法名都用了小寫駝峰因?yàn)?ArkTS 側(cè)解析 MethodCall 時(shí)方法名是直接字符串匹配風(fēng)格統(tǒng)一能減少低級(jí)錯(cuò)誤。invokeMethod內(nèi)部可以傳MapString, Object?但嵌套 map 的 value 類型在跨通道傳輸時(shí)會(huì)被序列化成 JSON所以messageAttributes這里我限制成MapString, String避免復(fù)雜結(jié)構(gòu)中int和double在 JSON 解析時(shí)的邊界問(wèn)題。3.3 ArkTS 側(cè)實(shí)現(xiàn)SigV4 簽名與 HTTPS 請(qǐng)求ArkTS 側(cè)的插件實(shí)現(xiàn)是整個(gè)適配的核心。首先要實(shí)現(xiàn) FlutterPlugin 接口在onAttachToFlutterEngine里注冊(cè) MethodCallHandler。import { FlutterPlugin } from ohos/flutter_plugin; import { MethodCall, MethodChannel } from ohos/flutter_plugin_bridge; import { http } from kit.NetworkKit; import { cryptoFramework } from kit.CryptoArchitectureKit; export class AwsSqsApiPlugin implements FlutterPlugin { private channel: MethodChannel | null null; onAttachToFlutterEngine(flutterEngine: any): void { this.channel new MethodChannel(flutterEngine, aws_sqs_api); this.channel.setMethodCallHandler((call: MethodCall) { return this.handleMethodCall(call); }); } private async handleMethodCall(call: MethodCall): Promiseany { const args call.arguments as Recordstring, Object; switch (call.method) { case sendMessage: return AwsSqsApi.sendMessage(args); case receiveMessage: return AwsSqsApi.receiveMessage(args); case deleteMessage: return AwsSqsApi.deleteMessage(args); case changeMessageVisibility: return AwsSqsApi.changeMessageVisibility(args); default: throw new Error(Unknown method: ${call.method}); } } onDetachFromFlutterEngine(flutterEngine: any): void { this.channel?.setMethodCallHandler(null); this.channel null; } }然后在AwsSqsApi類里實(shí)現(xiàn)具體的 SQS API 調(diào)用。這里最繞的是 SigV4 簽名我把它拆成了幾個(gè)工具方法。先看核心的簽名邏輯class AwsSqsApi { static async sendMessage(args: Recordstring, Object): Promisestring { const queueUrl args[queueUrl] as string; const messageBody args[messageBody] as string; const delaySeconds args[delaySeconds] as number; const messageAttributes args[messageAttributes] as Recordstring, string; const host extractHost(queueUrl); const region extractRegion(host); const payload buildPayload(messageBody, delaySeconds, messageAttributes); const signature await signRequest({ method: POST, host: host, path: /, query: , payload: payload, region: region, service: sqs, accessKey: CredentialManager.getAccessKey(), secretKey: CredentialManager.getSecretKey(), sessionToken: CredentialManager.getSessionToken(), }); const header http.HttpRequest; const request await http.createHttp().request(host, { method: http.RequestMethod.POST, header: { Content-Type: application/x-www-form-urlencoded, X-Amz-Date: signature.amzDate, Authorization: signature.authorization, X-Amz-Security-Token: CredentialManager.getSessionToken(), }, extraData: payload, expectDataType: http.HttpDataType.STRING, }); if (request.responseCode ! 200) { throw new Error(SQS request failed: ${request.responseCode} ${request.result}); } return parseMessageId(request.result); } }這里我對(duì)每一步展開(kāi)說(shuō)明。buildPayload會(huì)把 SQS 的請(qǐng)求參數(shù)拼成application/x-www-form-urlencoded格式這是 SQS REST API 的標(biāo)準(zhǔn)格式。實(shí)際的請(qǐng)求體長(zhǎng)這樣ActionSendMessageVersion2012-11-05QueueUrlhttps%3A%2F%2Fsqs.us-east-1.amazonaws.com%2F123456789012%2Fmy-queueMessageBodyhelloSigV4 簽名的計(jì)算我用的是cryptoFramework里的createMac接口做 HMAC-SHA256。核心步驟是async function signRequest(requestInfo: RequestInfo): PromiseSignature { const date new Date(); const amzDate formatAmzDate(date); const dateStamp formatDateStamp(date); const canonicalRequest buildCanonicalRequest(requestInfo.method, requestInfo.path, requestInfo.payload); const stringToSign AWS4-HMAC-SHA256\n${amzDate}\n${dateStamp}/${requestInfo.region}/sqs/aws4_request\n${sha256Hex(canonicalRequest)}; const kDate await hmacSha256(AWS4${requestInfo.secretKey}, dateStamp); const kRegion await hmacSha256(kDate, requestInfo.region); const kService await hmacSha256(kRegion, sqs); const kSigning await hmacSha256(kService, aws4_request); const signature await hmacSha256(kSigning, stringToSign); const credentialScope ${dateStamp}/${requestInfo.region}/sqs/aws4_request; const authorization AWS4-HMAC-SHA256 Credential${requestInfo.accessKey}/${credentialScope}, SignedHeaderscontent-type;host;x-amz-date, Signature${bytesToHex(signature)}; return { authorization, amzDate }; }寫這部分的時(shí)候我踩了一個(gè)很深的坑cryptoFramework的hmacSha256返回的是Uint8Array直接轉(zhuǎn)字符串會(huì)拿到亂碼必須先把 key 轉(zhuǎn)成Uint8Array再做二進(jìn)制拼接。上面代碼里hmacSha256(kDate, requestInfo.region)這里的kDate是上一輪的二進(jìn)制輸出不能直接toString()否則簽名結(jié)果永遠(yuǎn)和服務(wù)端對(duì)不上。4. 消費(fèi)者側(cè)的高可用設(shè)計(jì)4.1 可見(jiàn)性超時(shí)與消費(fèi)失敗處理消息發(fā)得出去不算完消費(fèi)端的高可用才是真正考驗(yàn)設(shè)計(jì)功底的地方。SQS 的消息模型是拉取后隱藏消費(fèi)者調(diào)用ReceiveMessage拿到消息后這條消息并不會(huì)立刻從隊(duì)列刪除而是進(jìn)入不可見(jiàn)狀態(tài)。這個(gè)不可見(jiàn)時(shí)間就叫 Visibility Timeout可見(jiàn)性超時(shí)。理解這個(gè)機(jī)制非常重要。如果消費(fèi)者在超時(shí)時(shí)間內(nèi)沒(méi)有調(diào)用DeleteMessage刪除消息SQS 會(huì)認(rèn)為消費(fèi)失敗把消息重新放回隊(duì)列再次對(duì)消費(fèi)者可見(jiàn)。這就像你從快遞柜取了個(gè)包裹但沒(méi)在時(shí)限內(nèi)拿走柜門會(huì)重新打開(kāi)包裹又變成待取狀態(tài)。OpenHarmony 客戶端上我設(shè)置的默認(rèn)可見(jiàn)性超時(shí)是 30 秒但實(shí)際業(yè)務(wù)處理完一條消息的平均耗時(shí)只有 2 到 3 秒。為什么留這么大的余量因?yàn)樵O(shè)備端的網(wǎng)絡(luò)狀況不穩(wěn)定弱網(wǎng)環(huán)境下 SQS 的響應(yīng)可能會(huì)延遲如果超時(shí)設(shè)得太短很容易造成消息在業(yè)務(wù)還沒(méi)處理完時(shí)就被重新推送導(dǎo)致重復(fù)消費(fèi)。如果超時(shí)設(shè)得太長(zhǎng)又要擔(dān)心消費(fèi)者崩潰后消息長(zhǎng)時(shí)間無(wú)人處理。我的處理策略是拉取到消息后立刻調(diào)用一次ChangeMessageVisibility把超時(shí)時(shí)間調(diào)整到 60 秒給業(yè)務(wù)處理預(yù)留充足時(shí)間業(yè)務(wù)處理成功后調(diào)用DeleteMessage刪除消息。如果業(yè)務(wù)處理失敗不調(diào)用刪除讓消息在超時(shí)后自動(dòng)回到隊(duì)列實(shí)現(xiàn)天然的重試機(jī)制。4.2 長(zhǎng)輪詢與批量拉取SQS 的消費(fèi)者如果頻繁輪詢空隊(duì)列會(huì)產(chǎn)生大量無(wú)效 API 調(diào)用既費(fèi)錢又費(fèi)電。Wi-Fi 環(huán)境下這個(gè)問(wèn)題不明顯但 OpenHarmony 設(shè)備往往是帶電池的功耗控制很關(guān)鍵。SQS 提供了長(zhǎng)輪詢機(jī)制在ReceiveMessage請(qǐng)求里帶WaitTimeSeconds參數(shù)可以設(shè)置 1 到 20 秒。當(dāng)隊(duì)列為空時(shí)請(qǐng)求不會(huì)立刻返回空列表而是掛住等待新消息到來(lái)或者直到超時(shí)時(shí)間結(jié)束。這樣消費(fèi)者每 20 秒只需要發(fā)起一次請(qǐng)求功耗大幅下降。批量拉取方面SQS 限制單次ReceiveMessage最多返回 10 條消息。我在 ArkTS 側(cè)做了循環(huán)拉取一次業(yè)務(wù)觸發(fā)最多拉取 50 條分 5 個(gè)批次并行處理每批之間加一個(gè) 100ms 的間隔避免瞬間打滿網(wǎng)絡(luò)帶寬。實(shí)測(cè)下來(lái)在 2000 條消息積壓的情況下消費(fèi)完所有消息只需要 4 秒左右。static async receiveBatch(queueUrl: string, visibilityTimeout: number, batchSize: number): PromiseListObject { const results: Object[] []; const batches Math.ceil(batchSize / 10); for (let i 0; i batches; i) { const receiveResult await this.receiveMessage({ queueUrl: queueUrl, maxNumberOfMessages: 10, waitTimeSeconds: 0, visibilityTimeout: visibilityTimeout, }); results.push(...receiveResult); if (receiveResult.length 10) { break; } await delay(100); } return results; }4.3 死信隊(duì)列兜底再穩(wěn)的系統(tǒng)也有處理不了的消息。比如設(shè)備上報(bào)了一條格式損壞的 JSON消費(fèi)程序每次解析都會(huì)失敗重試 10 次還是失敗。如果任由這種消息在隊(duì)列里反復(fù)橫跳不僅浪費(fèi)處理能力還會(huì)擠占正常消息的位置。SQS 的死信隊(duì)列DLQ就是干這個(gè)的。在主隊(duì)列的 Attributes 里配置 RedrivePolicy指定maxReceiveCount為 3 或 5這樣一條消息被拉取超過(guò)指定次數(shù)后SQS 會(huì)自動(dòng)把它轉(zhuǎn)移到對(duì)應(yīng)的死信隊(duì)列。死信隊(duì)列里的消息可以等開(kāi)發(fā)人員修復(fù) bug 后重新投遞回主隊(duì)列或者直接人工處理。在 OpenHarmony 客戶端的適配里我把死信隊(duì)列的消費(fèi)單獨(dú)做了一個(gè)通道。正常情況下客戶端只消費(fèi)主隊(duì)列死信隊(duì)列的消費(fèi)由后端來(lái)處理?？蛻舳税l(fā)現(xiàn)消息拉取次數(shù)異常時(shí)會(huì)記錄日志并上報(bào)一條告警事件方便運(yùn)維人員及時(shí)發(fā)現(xiàn)。5. 實(shí)測(cè)中的坑與排查技巧5.1 SignatureDoesNotMatch我排查了一天的簽名問(wèn)題這個(gè)錯(cuò)誤絕對(duì)是我這次適配里耗時(shí)最長(zhǎng)的問(wèn)題?，F(xiàn)象很簡(jiǎn)單在 Android 上跑得好好的同樣的參數(shù)搬到 OpenHarmony 上就報(bào)SignatureDoesNotMatch: The request signature we calculated does not match the signature you provided. Check your AWS Secret Access Key and signing method.我第一反應(yīng)是憑證問(wèn)題反復(fù)檢查了 AccessKey 和 SecretKey確認(rèn)沒(méi)問(wèn)題。然后又懷疑是 ArkTS 的 HMAC 實(shí)現(xiàn)有 bug打印出簽名值逐字節(jié)比對(duì)發(fā)現(xiàn)也沒(méi)有問(wèn)題。最后查到問(wèn)題出在 CanonicalRequest 里的 host 頭。Dart 的aws_signature_v4在簽名時(shí)用的是小寫 host比如sqs.us-east-1.amazonaws.com。但 ArkTS 的ohos.net.http在發(fā)送請(qǐng)求時(shí)某些版本會(huì)在 header 里自動(dòng)加上一個(gè)默認(rèn)的Host頭而且這個(gè) Host 頭的格式可能是SQS.US-EAST-1.AMAZONAWS.COM全大寫。SQS 服務(wù)端在驗(yàn)證簽名時(shí)是區(qū)分大小寫的host 頭不一致簽名自然對(duì)不上。解決方案是手動(dòng)設(shè)置請(qǐng)求的 header把 host 頭固定成小寫。還有一次是為了兼容簽名區(qū)域問(wèn)題改配置也排查了很久最后統(tǒng)一用us-east-1測(cè)試環(huán)境驗(yàn)證才定位到是區(qū)域參數(shù)傳遞錯(cuò)誤。這些都是第一線實(shí)操才會(huì)遇到的事。header: { Content-Type: application/x-www-form-urlencoded, Host: host.toLowerCase(), X-Amz-Date: signature.amzDate, Authorization: signature.authorization, }排查建議先在電腦上用 curl 模擬完整的 SQS 請(qǐng)求把簽名過(guò)程中每一步的中間值打印出來(lái)再用同樣的參數(shù)在 OpenHarmony 設(shè)備上跑對(duì)比兩個(gè)中間值哪里開(kāi)始不一致。這個(gè)方法我屢試不爽。5.2 消息積壓消費(fèi)者線程被系統(tǒng)掛起了OpenHarmony 對(duì)后臺(tái)任務(wù)的限制比 Android 更嚴(yán)格。應(yīng)用退到后臺(tái)后如果沒(méi)有任何前臺(tái)服務(wù)或長(zhǎng)時(shí)任務(wù)在運(yùn)行ArkTS 側(cè)執(zhí)行網(wǎng)絡(luò)請(qǐng)求的協(xié)程會(huì)在幾分鐘內(nèi)被系統(tǒng)掛起。表現(xiàn)就是應(yīng)用在后臺(tái)時(shí)消息不消費(fèi)回到前臺(tái)后突然開(kāi)始大量消費(fèi)積壓消息。解決思路有兩個(gè)方向我最終都做了在模塊的module.json5里聲明長(zhǎng)時(shí)任務(wù)權(quán)限參考常見(jiàn)鴻蒙適配方案申請(qǐng)后臺(tái)任務(wù)類型并配置對(duì)應(yīng)的權(quán)限這樣應(yīng)用在后臺(tái)運(yùn)行時(shí)有系統(tǒng)級(jí)別的資源保障。在 ArkTS 側(cè)用 WorkSchedulerExtension 定期喚醒每次喚醒拉取一批消息處理完再讓系統(tǒng)休眠。實(shí)測(cè)下來(lái)消息積壓時(shí)間窗口從原來(lái)的 10 分鐘以上控制到了 1 分鐘以內(nèi)。5.3 重復(fù)消費(fèi)正確使用 ReceiptHandleSQS 的消費(fèi)模型是 at-least-once也就是至少一次不保證恰好一次。重復(fù)消費(fèi)的根源在于網(wǎng)絡(luò)超時(shí)比如客戶端已經(jīng)調(diào)用了DeleteMessage但響應(yīng)在傳輸過(guò)程中丟失服務(wù)端沒(méi)收到刪除指令超時(shí)后消息再次變得可見(jiàn)。要減少重復(fù)消費(fèi)唯一可靠的手段是讓消費(fèi)邏輯冪等。我在設(shè)備端對(duì)每條消息計(jì)算了一個(gè)業(yè)務(wù)唯一 ID寫進(jìn)MessageAttributes的messageId字段。消費(fèi)端在處理前先查一下本地?cái)?shù)據(jù)庫(kù)如果這個(gè) ID 已經(jīng)處理過(guò)直接跳過(guò)。這個(gè)方案不能說(shuō) 100% 杜絕重復(fù)但能把影響降到可以忽略的程度。5.4 常見(jiàn)問(wèn)題速查表問(wèn)題現(xiàn)象可能原因排查方向SignatureDoesNotMatchhost 頭大小寫不一致檢查請(qǐng)求 header 中的 Host 是否為小寫AccessDenied憑證錯(cuò)誤或區(qū)域不匹配檢查 AccessKey/SecretKey確認(rèn) region 參數(shù)QueueDoesNotExistQueueUrl 填錯(cuò)或權(quán)限不足檢查 QueueUrl 的完整路徑確認(rèn)隊(duì)列和憑證歸屬同一賬號(hào)MethodChannel 調(diào)用超時(shí)ArkTS 側(cè)網(wǎng)絡(luò)請(qǐng)求阻塞檢查網(wǎng)絡(luò)權(quán)限確認(rèn)module.json5中已聲明 INTERNET消息積壓且應(yīng)用在后臺(tái)后臺(tái)任務(wù)被掛起配置長(zhǎng)時(shí)任務(wù)權(quán)限或使用 WorkSchedulerExtension后臺(tái)拿不到動(dòng)態(tài)憑證憑證刷新邏輯沒(méi)跑后臺(tái)在 ArkTS 側(cè)啟動(dòng)定時(shí)刷新保證 sessionToken 不過(guò)期5.5 憑證管理的安全實(shí)踐最后單獨(dú)聊聊憑證。AWS 的憑證如果寫死到客戶端里逆向出一個(gè)就能刷爆你的隊(duì)列。我在 ArkTS 側(cè)做了一層封裝憑證不會(huì)明文存儲(chǔ)在本地用系統(tǒng)的憑據(jù)加密能力加密后存入應(yīng)用沙箱。每次 Build 時(shí)從服務(wù)端拉取臨時(shí)憑證搭配 STS 的 sessionToken 使用過(guò)期后自動(dòng)刷新。這樣即使設(shè)備被 root泄露的也只是一段時(shí)間內(nèi)的臨時(shí)憑證影響范圍可控。6. 這套方案的后續(xù)擴(kuò)展方向把a(bǔ)ws_sqs_api在 OpenHarmony 上跑通不是終點(diǎn)它給后續(xù)的架構(gòu)演進(jìn)留了好幾個(gè)口子。一個(gè)是消息類型的擴(kuò)展。現(xiàn)在發(fā)送的消息體是普通字符串但 SQS 的MessageAttributes支持結(jié)構(gòu)化屬性可以在發(fā)送時(shí)打上設(shè)備類型、環(huán)境、業(yè)務(wù)標(biāo)簽消費(fèi)端根據(jù)這些屬性做路由和處理策略分流。另一個(gè)是隊(duì)列策略的調(diào)整。SQS 有個(gè)延時(shí)隊(duì)列功能可以把消息延遲 0 到 900 秒后再對(duì)消費(fèi)者可見(jiàn)。這個(gè)能力可以用來(lái)做設(shè)備升級(jí)的時(shí)間窗口控制比如設(shè)備收到升級(jí)指令后不用立刻執(zhí)行而是先把指令投遞到延遲隊(duì)列過(guò) 15 分鐘再拉取執(zhí)行避開(kāi)業(yè)務(wù)高峰。最后說(shuō)一句實(shí)在話OpenHarmony 的 Flutter 生態(tài)還在快速完善階段很多在三方庫(kù)上的適配工作沒(méi)有太多現(xiàn)成資料可查。遇到問(wèn)題多看官方文檔、多打印日志、多跟同類項(xiàng)目的開(kāi)發(fā)者交流比自己悶頭排查高效得多。這篇復(fù)盤里寫的坑都是我實(shí)實(shí)在在踩過(guò)的能幫你少走幾步彎路就是它最大的價(jià)值。 OpenHarmony上Flutter插件aws_sqs_api適配實(shí)戰(zhàn)
去年接了個(gè)挺頭疼的活把公司一套基于 Flutter 的客戶端應(yīng)用遷移到 OpenHarmony 設(shè)備上。界面、狀態(tài)管理、本地存儲(chǔ)都順利解決了最后卡在一個(gè)叫aws_sqs_api的三方庫(kù)上。這個(gè)庫(kù)是 AWS SQSSimple Queue Service的 Dart 客戶端我們?cè)谠?Android/iOS 版本里用它做設(shè)備端數(shù)據(jù)上報(bào)把采集到的狀態(tài)、日志、業(yè)務(wù)事件一股腦丟進(jìn)云端隊(duì)列后端服務(wù)異步消費(fèi)實(shí)現(xiàn)分布式場(chǎng)景下的消息解耦和削峰填谷。搬到 OpenHarmony 后這個(gè)鏈路必須原樣跑通否則所有設(shè)備上報(bào)都會(huì)變成直連后端 HTTP 接口一旦流量抖動(dòng)后端就會(huì)被打爆。這篇文章就是那次完整適配過(guò)程的復(fù)盤里面包含方案取舍、MethodChannel 橋接細(xì)節(jié)、SigV4 簽名在 ArkTS 側(cè)的實(shí)現(xiàn)以及我在生產(chǎn)環(huán)境里踩過(guò)的坑希望能給同樣在 OpenHarmony 上做 Flutter 插件適配的同學(xué)省點(diǎn)時(shí)間。1. 這個(gè)庫(kù)到底是干什么的分布式消息異步解耦的切入點(diǎn)1.1 為什么選 SQS 而不是其他消息中間件先聊聊背景。我們的場(chǎng)景是物聯(lián)網(wǎng)設(shè)備端上報(bào)設(shè)備數(shù)量上千臺(tái)每臺(tái)每隔幾秒就會(huì)產(chǎn)生一條狀態(tài)數(shù)據(jù)。如果設(shè)備直連后端 API高峰期每秒可能有上千個(gè)并發(fā)請(qǐng)求后端服務(wù)要么瘋狂擴(kuò)容要么直接限流丟數(shù)據(jù)。用消息隊(duì)列做中轉(zhuǎn)以后設(shè)備端只負(fù)責(zé)把消息丟進(jìn)隊(duì)列后端按自己的處理能力去拉取兩端互不阻塞這就是典型的異步解耦。技術(shù)選型時(shí)我們對(duì)比過(guò) RabbitMQ、Kafka 和 AWS SQS。自建 RabbitMQ 或 Kafka 在云端要考慮運(yùn)維成本而且我們的客戶端是 Flutter 寫的需要找 Dart 生態(tài)里維護(hù)活躍的 SDK。AWS SQS 雖然是云廠商托管服務(wù)但勝在完全不用運(yùn)維標(biāo)準(zhǔn)隊(duì)列無(wú)限吞吐還有死信隊(duì)列、延遲隊(duì)列、長(zhǎng)輪詢這些開(kāi)箱即用的能力。配合aws_sqs_api這個(gè)純 Dart 包Dart 層直接調(diào)用 SQS 的 REST API省掉了中間再套一層自建網(wǎng)關(guān)的成本。1.2 aws_sqs_api 的功能邊界與依賴關(guān)系aws_sqs_api是 AWS 官方為 Dart 語(yǔ)言生成的 SQS API 客戶端底層用的是 AWS 的 Smithy 代碼生成框架。它本身不包含 UI 組件也不依賴任何 Flutter 原生插件核心能力就是封裝 SQS 的 REST API 調(diào)用包括創(chuàng)建隊(duì)列、發(fā)送消息、接收消息、刪除消息、修改可見(jiàn)性超時(shí)等。它有幾個(gè)關(guān)鍵依賴包需要一起引入aws_common提供 AWS 服務(wù)的通用基礎(chǔ)類型和配置aws_signature_v4實(shí)現(xiàn) AWS Signature Version 4 請(qǐng)求簽名aws_smithy_clientSmithy 客戶端運(yùn)行時(shí)負(fù)責(zé) HTTP 請(qǐng)求的發(fā)送和響應(yīng)解析這個(gè)依賴關(guān)系很重要。aws_signature_v4是純 Dart 實(shí)現(xiàn)的簽名算法理論上在任何能跑 Dart 的平臺(tái)上都能運(yùn)行。但實(shí)際適配 OpenHarmony 時(shí)問(wèn)題往往出在更底層——Dart 運(yùn)行時(shí)能不能正常發(fā) HTTPS 請(qǐng)求、網(wǎng)絡(luò)權(quán)限怎么配、憑證存哪里。搞清楚這些邊界你就知道鴻蒙適配的重點(diǎn)其實(shí)不在 Dart 層而在平臺(tái)橋接層。2. 鴻蒙適配的核心難點(diǎn)拆解2.1 三層問(wèn)題運(yùn)行時(shí)、簽名、原生通道把a(bǔ)ws_sqs_api搬到 OpenHarmony我把它拆成了三個(gè)層次的問(wèn)題逐個(gè)擊破第一層是 Dart 運(yùn)行時(shí)兼容性。OpenHarmony 上的 Flutter 是基于 OpenHarmony 官方移植的 Flutter SDK 來(lái)跑的大部分dart:io的能力都支持但跟 Android/iOS 的 Flutter 運(yùn)行時(shí)不是同一個(gè)實(shí)現(xiàn)。我們?cè)谶m配過(guò)程中發(fā)現(xiàn)aws_smithy_client里的某些網(wǎng)絡(luò)異常處理在 OpenHarmony 上的表現(xiàn)略有差異具體來(lái)說(shuō)是SocketException的報(bào)錯(cuò)信息格式不一樣導(dǎo)致日志解析邏輯需要微調(diào)。這一層的問(wèn)題比較隱蔽建議適配時(shí)先寫一個(gè)最小化的 Dart 腳本在目標(biāo)設(shè)備上跑一遍確認(rèn) HttpClient 能正常訪問(wèn)外網(wǎng)。第二層是 SigV4 簽名算法的平臺(tái)差異。aws_signature_v4用的是純 Dart 的crypto包做 SHA256 和 HMAC 計(jì)算在 OpenHarmony 的 Dart 運(yùn)行時(shí)上可以正常工作。但這里有個(gè)坑SQS 的請(qǐng)求簽名要求 CanonicalRequest 里的 host 頭必須和實(shí)際請(qǐng)求的 host 完全一致包括大小寫和端口號(hào)。在 OpenHarmony 上如果你走了代理或者自定義了網(wǎng)絡(luò)棧host 頭可能會(huì)被改寫導(dǎo)致服務(wù)端返回SignatureDoesNotMatch。第三層是原生平臺(tái)通道。項(xiàng)目里的憑證信息之前是存在系統(tǒng)鑰匙串里的Android 用的是flutter_secure_storageiOS 用 Keychain。OpenHarmony 上沒(méi)有現(xiàn)成的插件這層必須自己寫原生橋接。另外我們的業(yè)務(wù)還要求 App 在后臺(tái)時(shí)也能持續(xù)消費(fèi)隊(duì)列消息這需要鴻蒙端的任務(wù)后臺(tái)執(zhí)行能力配合不是一個(gè)純 Dart 包能解決的。所以適配工作的重心最終落在了 MethodChannel 的建聯(lián)和 ArkTS 原生側(cè)的實(shí)現(xiàn)上。2.2 MethodChannel 橋接 vs 純 Dart 直連的取舍有一種思路是既然aws_sqs_api是純 Dart 包OpenHarmony 的 Flutter 運(yùn)行時(shí)又支持dart:io那是不是什么都不用改直接跑就完事了我最初也是這么想的在開(kāi)發(fā)機(jī)上跑了個(gè) demo還真能通。但放到生產(chǎn)環(huán)境就暴露了三個(gè)問(wèn)題憑證存儲(chǔ)沒(méi)有安全的地方。Dart 側(cè)只能用 shared_preferences 之類的插件存明文這在合規(guī)審計(jì)上過(guò)不去。后臺(tái)消費(fèi)不可靠。Flutter 的 Dart isolate 在應(yīng)用退到后臺(tái)后可能被系統(tǒng)掛起沒(méi)有鴻蒙端原生任務(wù)配合消息消費(fèi)會(huì)斷。網(wǎng)絡(luò)棧不可控。某些定制 ROM 的 OpenHarmony 設(shè)備會(huì)對(duì) Flutter 的 HttpClient 做限制而走系統(tǒng)ohos.net.http是經(jīng)過(guò)充分驗(yàn)證的通道。所以最終方案是Dart 層通過(guò) MethodChannel 調(diào) ArkTS 原生實(shí)現(xiàn)把發(fā)送消息、接收消息、刪除消息、修改可見(jiàn)性這幾個(gè)核心操作全部下沉到鴻蒙側(cè)。aws_sqs_api在 Dart 層保留作為接口定義和數(shù)據(jù)模型參考真正發(fā) HTTP 請(qǐng)求的是 ArkTS 代碼。這個(gè)方案雖然多寫了不少原生代碼但換來(lái)的是安全存儲(chǔ)、穩(wěn)定網(wǎng)絡(luò)和后臺(tái)執(zhí)行能力我認(rèn)為是值得的。3. 實(shí)操?gòu)膭?chuàng)建插件工程到跑通第一條消息3.1 工程結(jié)構(gòu)配置與權(quán)限聲明OpenHarmony 的 Flutter 插件和 Android 插件結(jié)構(gòu)很像但目錄名從android換成了ohos。我用的是手動(dòng)創(chuàng)建的方式因?yàn)閒lutter create --templateplugin默認(rèn)不支持生成 ohos 目錄。工程目錄結(jié)構(gòu)長(zhǎng)這樣aws_sqs_api_ohos/ ├── pubspec.yaml ├── lib/ │ ├── aws_sqs_api_ohos.dart │ └── src/ │ └── (dart層方法通道封裝) └── ohos/ ├── build-profile.json5 └── entry/ └── src/ └── main/ ├── ets/ │ ├── entryability/ │ └── plugins/ │ └── AwsSqsApiPlugin.ets └── module.json5pubspec.yaml里要聲明插件支持的平臺(tái)注意要加上 ohosflutter: plugin: platforms: android: package: com.example.aws_sqs_api_ohos pluginClass: AwsSqsApiPlugin ios: pluginClass: AwsSqsApiPlugin ohos: pluginClass: AwsSqsApiPlugin pluginImplementation: AwsSqsApiPluginImplmodule.json5里必須聲明網(wǎng)絡(luò)權(quán)限這是最容易漏的一步。鴻蒙應(yīng)用默認(rèn)沒(méi)有網(wǎng)絡(luò)訪問(wèn)權(quán)限不加這個(gè)權(quán)限所有 HTTPS 請(qǐng)求都會(huì)靜默失敗{ module: { name: entry, requestPermissions: [ { name: ohos.permission.INTERNET } ] } }這個(gè)權(quán)限配置和 Android 的AndroidManifest.xml里加uses-permission android:nameandroid.permission.INTERNET /是同一個(gè)作用但位置完全不同很多從 Android 轉(zhuǎn)過(guò)來(lái)的同學(xué)會(huì)下意識(shí)去找 manifest 文件結(jié)果在鴻蒙工程里根本找不到。3.2 Dart 側(cè)封裝MethodChannel 的調(diào)用契約Dart 側(cè)的封裝盡量保持和原來(lái)aws_sqs_api的調(diào)用風(fēng)格一致這樣業(yè)務(wù)代碼不用大面積改動(dòng)。我定義了一個(gè)統(tǒng)一的方法通道名aws_sqs_api然后按 SQS 的核心操作拆成幾個(gè)方法。class AwsSqsApiOhos { static const MethodChannel _channel MethodChannel(aws_sqs_api); static FutureString sendMessage({ required String queueUrl, required String messageBody, int delaySeconds 0, MapString, String attributes const {}, }) async { final result await _channel.invokeMethod(sendMessage, { queueUrl: queueUrl, messageBody: messageBody, delaySeconds: delaySeconds, messageAttributes: attributes, }); return result as String; } static FutureListMapString, dynamic receiveMessage({ required String queueUrl, int maxNumberOfMessages 10, int waitTimeSeconds 0, int visibilityTimeout 30, }) async { final result await _channel.invokeMethod(receiveMessage, { queueUrl: queueUrl, maxNumberOfMessages: maxNumberOfMessages, waitTimeSeconds: waitTimeSeconds, visibilityTimeout: visibilityTimeout, }); return (result as List).castMapString, dynamic(); } static Futurebool deleteMessage({ required String queueUrl, required String receiptHandle, }) async { final result await _channel.invokeMethod(deleteMessage, { queueUrl: queueUrl, receiptHandle: receiptHandle, }); return result as bool; } static Futurevoid changeMessageVisibility({ required String queueUrl, required String receiptHandle, required int visibilityTimeout, }) async { await _channel.invokeMethod(changeMessageVisibility, { queueUrl: queueUrl, receiptHandle: receiptHandle, visibilityTimeout: visibilityTimeout, }); } }注意幾個(gè)設(shè)計(jì)細(xì)節(jié)receiveMessage的返回值我用了ListMapString, dynamic而不是強(qiáng)類型對(duì)象因?yàn)?MethodChannel 的 JSON 反序列化在鴻蒙端的Map鍵值類型可能和 Dart 側(cè)不完全匹配留一層動(dòng)態(tài)類型可以減少類型轉(zhuǎn)換異常。所有方法名都用了小寫駝峰因?yàn)?ArkTS 側(cè)解析 MethodCall 時(shí)方法名是直接字符串匹配風(fēng)格統(tǒng)一能減少低級(jí)錯(cuò)誤。invokeMethod內(nèi)部可以傳MapString, Object?但嵌套 map 的 value 類型在跨通道傳輸時(shí)會(huì)被序列化成 JSON所以messageAttributes這里我限制成MapString, String避免復(fù)雜結(jié)構(gòu)中int和double在 JSON 解析時(shí)的邊界問(wèn)題。3.3 ArkTS 側(cè)實(shí)現(xiàn)SigV4 簽名與 HTTPS 請(qǐng)求ArkTS 側(cè)的插件實(shí)現(xiàn)是整個(gè)適配的核心。首先要實(shí)現(xiàn) FlutterPlugin 接口在onAttachToFlutterEngine里注冊(cè) MethodCallHandler。import { FlutterPlugin } from ohos/flutter_plugin; import { MethodCall, MethodChannel } from ohos/flutter_plugin_bridge; import { http } from kit.NetworkKit; import { cryptoFramework } from kit.CryptoArchitectureKit; export class AwsSqsApiPlugin implements FlutterPlugin { private channel: MethodChannel | null null; onAttachToFlutterEngine(flutterEngine: any): void { this.channel new MethodChannel(flutterEngine, aws_sqs_api); this.channel.setMethodCallHandler((call: MethodCall) { return this.handleMethodCall(call); }); } private async handleMethodCall(call: MethodCall): Promiseany { const args call.arguments as Recordstring, Object; switch (call.method) { case sendMessage: return AwsSqsApi.sendMessage(args); case receiveMessage: return AwsSqsApi.receiveMessage(args); case deleteMessage: return AwsSqsApi.deleteMessage(args); case changeMessageVisibility: return AwsSqsApi.changeMessageVisibility(args); default: throw new Error(Unknown method: ${call.method}); } } onDetachFromFlutterEngine(flutterEngine: any): void { this.channel?.setMethodCallHandler(null); this.channel null; } }然后在AwsSqsApi類里實(shí)現(xiàn)具體的 SQS API 調(diào)用。這里最繞的是 SigV4 簽名我把它拆成了幾個(gè)工具方法。先看核心的簽名邏輯class AwsSqsApi { static async sendMessage(args: Recordstring, Object): Promisestring { const queueUrl args[queueUrl] as string; const messageBody args[messageBody] as string; const delaySeconds args[delaySeconds] as number; const messageAttributes args[messageAttributes] as Recordstring, string; const host extractHost(queueUrl); const region extractRegion(host); const payload buildPayload(messageBody, delaySeconds, messageAttributes); const signature await signRequest({ method: POST, host: host, path: /, query: , payload: payload, region: region, service: sqs, accessKey: CredentialManager.getAccessKey(), secretKey: CredentialManager.getSecretKey(), sessionToken: CredentialManager.getSessionToken(), }); const header http.HttpRequest; const request await http.createHttp().request(host, { method: http.RequestMethod.POST, header: { Content-Type: application/x-www-form-urlencoded, X-Amz-Date: signature.amzDate, Authorization: signature.authorization, X-Amz-Security-Token: CredentialManager.getSessionToken(), }, extraData: payload, expectDataType: http.HttpDataType.STRING, }); if (request.responseCode ! 200) { throw new Error(SQS request failed: ${request.responseCode} ${request.result}); } return parseMessageId(request.result); } }這里我對(duì)每一步展開(kāi)說(shuō)明。buildPayload會(huì)把 SQS 的請(qǐng)求參數(shù)拼成application/x-www-form-urlencoded格式這是 SQS REST API 的標(biāo)準(zhǔn)格式。實(shí)際的請(qǐng)求體長(zhǎng)這樣ActionSendMessageVersion2012-11-05QueueUrlhttps%3A%2F%2Fsqs.us-east-1.amazonaws.com%2F123456789012%2Fmy-queueMessageBodyhelloSigV4 簽名的計(jì)算我用的是cryptoFramework里的createMac接口做 HMAC-SHA256。核心步驟是async function signRequest(requestInfo: RequestInfo): PromiseSignature { const date new Date(); const amzDate formatAmzDate(date); const dateStamp formatDateStamp(date); const canonicalRequest buildCanonicalRequest(requestInfo.method, requestInfo.path, requestInfo.payload); const stringToSign AWS4-HMAC-SHA256\n${amzDate}\n${dateStamp}/${requestInfo.region}/sqs/aws4_request\n${sha256Hex(canonicalRequest)}; const kDate await hmacSha256(AWS4${requestInfo.secretKey}, dateStamp); const kRegion await hmacSha256(kDate, requestInfo.region); const kService await hmacSha256(kRegion, sqs); const kSigning await hmacSha256(kService, aws4_request); const signature await hmacSha256(kSigning, stringToSign); const credentialScope ${dateStamp}/${requestInfo.region}/sqs/aws4_request; const authorization AWS4-HMAC-SHA256 Credential${requestInfo.accessKey}/${credentialScope}, SignedHeaderscontent-type;host;x-amz-date, Signature${bytesToHex(signature)}; return { authorization, amzDate }; }寫這部分的時(shí)候我踩了一個(gè)很深的坑cryptoFramework的hmacSha256返回的是Uint8Array直接轉(zhuǎn)字符串會(huì)拿到亂碼必須先把 key 轉(zhuǎn)成Uint8Array再做二進(jìn)制拼接。上面代碼里hmacSha256(kDate, requestInfo.region)這里的kDate是上一輪的二進(jìn)制輸出不能直接toString()否則簽名結(jié)果永遠(yuǎn)和服務(wù)端對(duì)不上。4. 消費(fèi)者側(cè)的高可用設(shè)計(jì)4.1 可見(jiàn)性超時(shí)與消費(fèi)失敗處理消息發(fā)得出去不算完消費(fèi)端的高可用才是真正考驗(yàn)設(shè)計(jì)功底的地方。SQS 的消息模型是拉取后隱藏消費(fèi)者調(diào)用ReceiveMessage拿到消息后這條消息并不會(huì)立刻從隊(duì)列刪除而是進(jìn)入不可見(jiàn)狀態(tài)。這個(gè)不可見(jiàn)時(shí)間就叫 Visibility Timeout可見(jiàn)性超時(shí)。理解這個(gè)機(jī)制非常重要。如果消費(fèi)者在超時(shí)時(shí)間內(nèi)沒(méi)有調(diào)用DeleteMessage刪除消息SQS 會(huì)認(rèn)為消費(fèi)失敗把消息重新放回隊(duì)列再次對(duì)消費(fèi)者可見(jiàn)。這就像你從快遞柜取了個(gè)包裹但沒(méi)在時(shí)限內(nèi)拿走柜門會(huì)重新打開(kāi)包裹又變成待取狀態(tài)。OpenHarmony 客戶端上我設(shè)置的默認(rèn)可見(jiàn)性超時(shí)是 30 秒但實(shí)際業(yè)務(wù)處理完一條消息的平均耗時(shí)只有 2 到 3 秒。為什么留這么大的余量因?yàn)樵O(shè)備端的網(wǎng)絡(luò)狀況不穩(wěn)定弱網(wǎng)環(huán)境下 SQS 的響應(yīng)可能會(huì)延遲如果超時(shí)設(shè)得太短很容易造成消息在業(yè)務(wù)還沒(méi)處理完時(shí)就被重新推送導(dǎo)致重復(fù)消費(fèi)。如果超時(shí)設(shè)得太長(zhǎng)又要擔(dān)心消費(fèi)者崩潰后消息長(zhǎng)時(shí)間無(wú)人處理。我的處理策略是拉取到消息后立刻調(diào)用一次ChangeMessageVisibility把超時(shí)時(shí)間調(diào)整到 60 秒給業(yè)務(wù)處理預(yù)留充足時(shí)間業(yè)務(wù)處理成功后調(diào)用DeleteMessage刪除消息。如果業(yè)務(wù)處理失敗不調(diào)用刪除讓消息在超時(shí)后自動(dòng)回到隊(duì)列實(shí)現(xiàn)天然的重試機(jī)制。4.2 長(zhǎng)輪詢與批量拉取SQS 的消費(fèi)者如果頻繁輪詢空隊(duì)列會(huì)產(chǎn)生大量無(wú)效 API 調(diào)用既費(fèi)錢又費(fèi)電。Wi-Fi 環(huán)境下這個(gè)問(wèn)題不明顯但 OpenHarmony 設(shè)備往往是帶電池的功耗控制很關(guān)鍵。SQS 提供了長(zhǎng)輪詢機(jī)制在ReceiveMessage請(qǐng)求里帶WaitTimeSeconds參數(shù)可以設(shè)置 1 到 20 秒。當(dāng)隊(duì)列為空時(shí)請(qǐng)求不會(huì)立刻返回空列表而是掛住等待新消息到來(lái)或者直到超時(shí)時(shí)間結(jié)束。這樣消費(fèi)者每 20 秒只需要發(fā)起一次請(qǐng)求功耗大幅下降。批量拉取方面SQS 限制單次ReceiveMessage最多返回 10 條消息。我在 ArkTS 側(cè)做了循環(huán)拉取一次業(yè)務(wù)觸發(fā)最多拉取 50 條分 5 個(gè)批次并行處理每批之間加一個(gè) 100ms 的間隔避免瞬間打滿網(wǎng)絡(luò)帶寬。實(shí)測(cè)下來(lái)在 2000 條消息積壓的情況下消費(fèi)完所有消息只需要 4 秒左右。static async receiveBatch(queueUrl: string, visibilityTimeout: number, batchSize: number): PromiseListObject { const results: Object[] []; const batches Math.ceil(batchSize / 10); for (let i 0; i batches; i) { const receiveResult await this.receiveMessage({ queueUrl: queueUrl, maxNumberOfMessages: 10, waitTimeSeconds: 0, visibilityTimeout: visibilityTimeout, }); results.push(...receiveResult); if (receiveResult.length 10) { break; } await delay(100); } return results; }4.3 死信隊(duì)列兜底再穩(wěn)的系統(tǒng)也有處理不了的消息。比如設(shè)備上報(bào)了一條格式損壞的 JSON消費(fèi)程序每次解析都會(huì)失敗重試 10 次還是失敗。如果任由這種消息在隊(duì)列里反復(fù)橫跳不僅浪費(fèi)處理能力還會(huì)擠占正常消息的位置。SQS 的死信隊(duì)列DLQ就是干這個(gè)的。在主隊(duì)列的 Attributes 里配置 RedrivePolicy指定maxReceiveCount為 3 或 5這樣一條消息被拉取超過(guò)指定次數(shù)后SQS 會(huì)自動(dòng)把它轉(zhuǎn)移到對(duì)應(yīng)的死信隊(duì)列。死信隊(duì)列里的消息可以等開(kāi)發(fā)人員修復(fù) bug 后重新投遞回主隊(duì)列或者直接人工處理。在 OpenHarmony 客戶端的適配里我把死信隊(duì)列的消費(fèi)單獨(dú)做了一個(gè)通道。正常情況下客戶端只消費(fèi)主隊(duì)列死信隊(duì)列的消費(fèi)由后端來(lái)處理?？蛻舳税l(fā)現(xiàn)消息拉取次數(shù)異常時(shí)會(huì)記錄日志并上報(bào)一條告警事件方便運(yùn)維人員及時(shí)發(fā)現(xiàn)。5. 實(shí)測(cè)中的坑與排查技巧5.1 SignatureDoesNotMatch我排查了一天的簽名問(wèn)題這個(gè)錯(cuò)誤絕對(duì)是我這次適配里耗時(shí)最長(zhǎng)的問(wèn)題?，F(xiàn)象很簡(jiǎn)單在 Android 上跑得好好的同樣的參數(shù)搬到 OpenHarmony 上就報(bào)SignatureDoesNotMatch: The request signature we calculated does not match the signature you provided. Check your AWS Secret Access Key and signing method.我第一反應(yīng)是憑證問(wèn)題反復(fù)檢查了 AccessKey 和 SecretKey確認(rèn)沒(méi)問(wèn)題。然后又懷疑是 ArkTS 的 HMAC 實(shí)現(xiàn)有 bug打印出簽名值逐字節(jié)比對(duì)發(fā)現(xiàn)也沒(méi)有問(wèn)題。最后查到問(wèn)題出在 CanonicalRequest 里的 host 頭。Dart 的aws_signature_v4在簽名時(shí)用的是小寫 host比如sqs.us-east-1.amazonaws.com。但 ArkTS 的ohos.net.http在發(fā)送請(qǐng)求時(shí)某些版本會(huì)在 header 里自動(dòng)加上一個(gè)默認(rèn)的Host頭而且這個(gè) Host 頭的格式可能是SQS.US-EAST-1.AMAZONAWS.COM全大寫。SQS 服務(wù)端在驗(yàn)證簽名時(shí)是區(qū)分大小寫的host 頭不一致簽名自然對(duì)不上。解決方案是手動(dòng)設(shè)置請(qǐng)求的 header把 host 頭固定成小寫。還有一次是為了兼容簽名區(qū)域問(wèn)題改配置也排查了很久最后統(tǒng)一用us-east-1測(cè)試環(huán)境驗(yàn)證才定位到是區(qū)域參數(shù)傳遞錯(cuò)誤。這些都是第一線實(shí)操才會(huì)遇到的事。header: { Content-Type: application/x-www-form-urlencoded, Host: host.toLowerCase(), X-Amz-Date: signature.amzDate, Authorization: signature.authorization, }排查建議先在電腦上用 curl 模擬完整的 SQS 請(qǐng)求把簽名過(guò)程中每一步的中間值打印出來(lái)再用同樣的參數(shù)在 OpenHarmony 設(shè)備上跑對(duì)比兩個(gè)中間值哪里開(kāi)始不一致。這個(gè)方法我屢試不爽。5.2 消息積壓消費(fèi)者線程被系統(tǒng)掛起了OpenHarmony 對(duì)后臺(tái)任務(wù)的限制比 Android 更嚴(yán)格。應(yīng)用退到后臺(tái)后如果沒(méi)有任何前臺(tái)服務(wù)或長(zhǎng)時(shí)任務(wù)在運(yùn)行ArkTS 側(cè)執(zhí)行網(wǎng)絡(luò)請(qǐng)求的協(xié)程會(huì)在幾分鐘內(nèi)被系統(tǒng)掛起。表現(xiàn)就是應(yīng)用在后臺(tái)時(shí)消息不消費(fèi)回到前臺(tái)后突然開(kāi)始大量消費(fèi)積壓消息。解決思路有兩個(gè)方向我最終都做了在模塊的module.json5里聲明長(zhǎng)時(shí)任務(wù)權(quán)限參考常見(jiàn)鴻蒙適配方案申請(qǐng)后臺(tái)任務(wù)類型并配置對(duì)應(yīng)的權(quán)限這樣應(yīng)用在后臺(tái)運(yùn)行時(shí)有系統(tǒng)級(jí)別的資源保障。在 ArkTS 側(cè)用 WorkSchedulerExtension 定期喚醒每次喚醒拉取一批消息處理完再讓系統(tǒng)休眠。實(shí)測(cè)下來(lái)消息積壓時(shí)間窗口從原來(lái)的 10 分鐘以上控制到了 1 分鐘以內(nèi)。5.3 重復(fù)消費(fèi)正確使用 ReceiptHandleSQS 的消費(fèi)模型是 at-least-once也就是至少一次不保證恰好一次。重復(fù)消費(fèi)的根源在于網(wǎng)絡(luò)超時(shí)比如客戶端已經(jīng)調(diào)用了DeleteMessage但響應(yīng)在傳輸過(guò)程中丟失服務(wù)端沒(méi)收到刪除指令超時(shí)后消息再次變得可見(jiàn)。要減少重復(fù)消費(fèi)唯一可靠的手段是讓消費(fèi)邏輯冪等。我在設(shè)備端對(duì)每條消息計(jì)算了一個(gè)業(yè)務(wù)唯一 ID寫進(jìn)MessageAttributes的messageId字段。消費(fèi)端在處理前先查一下本地?cái)?shù)據(jù)庫(kù)如果這個(gè) ID 已經(jīng)處理過(guò)直接跳過(guò)。這個(gè)方案不能說(shuō) 100% 杜絕重復(fù)但能把影響降到可以忽略的程度。5.4 常見(jiàn)問(wèn)題速查表問(wèn)題現(xiàn)象可能原因排查方向SignatureDoesNotMatchhost 頭大小寫不一致檢查請(qǐng)求 header 中的 Host 是否為小寫AccessDenied憑證錯(cuò)誤或區(qū)域不匹配檢查 AccessKey/SecretKey確認(rèn) region 參數(shù)QueueDoesNotExistQueueUrl 填錯(cuò)或權(quán)限不足檢查 QueueUrl 的完整路徑確認(rèn)隊(duì)列和憑證歸屬同一賬號(hào)MethodChannel 調(diào)用超時(shí)ArkTS 側(cè)網(wǎng)絡(luò)請(qǐng)求阻塞檢查網(wǎng)絡(luò)權(quán)限確認(rèn)module.json5中已聲明 INTERNET消息積壓且應(yīng)用在后臺(tái)后臺(tái)任務(wù)被掛起配置長(zhǎng)時(shí)任務(wù)權(quán)限或使用 WorkSchedulerExtension后臺(tái)拿不到動(dòng)態(tài)憑證憑證刷新邏輯沒(méi)跑后臺(tái)在 ArkTS 側(cè)啟動(dòng)定時(shí)刷新保證 sessionToken 不過(guò)期5.5 憑證管理的安全實(shí)踐最后單獨(dú)聊聊憑證。AWS 的憑證如果寫死到客戶端里逆向出一個(gè)就能刷爆你的隊(duì)列。我在 ArkTS 側(cè)做了一層封裝憑證不會(huì)明文存儲(chǔ)在本地用系統(tǒng)的憑據(jù)加密能力加密后存入應(yīng)用沙箱。每次 Build 時(shí)從服務(wù)端拉取臨時(shí)憑證搭配 STS 的 sessionToken 使用過(guò)期后自動(dòng)刷新。這樣即使設(shè)備被 root泄露的也只是一段時(shí)間內(nèi)的臨時(shí)憑證影響范圍可控。6. 這套方案的后續(xù)擴(kuò)展方向把a(bǔ)ws_sqs_api在 OpenHarmony 上跑通不是終點(diǎn)它給后續(xù)的架構(gòu)演進(jìn)留了好幾個(gè)口子。一個(gè)是消息類型的擴(kuò)展。現(xiàn)在發(fā)送的消息體是普通字符串但 SQS 的MessageAttributes支持結(jié)構(gòu)化屬性可以在發(fā)送時(shí)打上設(shè)備類型、環(huán)境、業(yè)務(wù)標(biāo)簽消費(fèi)端根據(jù)這些屬性做路由和處理策略分流。另一個(gè)是隊(duì)列策略的調(diào)整。SQS 有個(gè)延時(shí)隊(duì)列功能可以把消息延遲 0 到 900 秒后再對(duì)消費(fèi)者可見(jiàn)。這個(gè)能力可以用來(lái)做設(shè)備升級(jí)的時(shí)間窗口控制比如設(shè)備收到升級(jí)指令后不用立刻執(zhí)行而是先把指令投遞到延遲隊(duì)列過(guò) 15 分鐘再拉取執(zhí)行避開(kāi)業(yè)務(wù)高峰。最后說(shuō)一句實(shí)在話OpenHarmony 的 Flutter 生態(tài)還在快速完善階段很多在三方庫(kù)上的適配工作沒(méi)有太多現(xiàn)成資料可查。遇到問(wèn)題多看官方文檔、多打印日志、多跟同類項(xiàng)目的開(kāi)發(fā)者交流比自己悶頭排查高效得多。這篇復(fù)盤里寫的坑都是我實(shí)實(shí)在在踩過(guò)的能幫你少走幾步彎路就是它最大的價(jià)值。