去年接了个挺头疼的活把公司一套基于 Flutter 的客户端应用迁移到 OpenHarmony 设备上。界面、状态管理、本地存储都顺利解决了最后卡在一个叫aws_sqs_api的三方库上。这个库是 AWS SQSSimple Queue Service的 Dart 客户端我们在原有 Android/iOS 版本里用它做设备端数据上报把采集到的状态、日志、业务事件一股脑丢进云端队列后端服务异步消费实现分布式场景下的消息解耦和削峰填谷。搬到 OpenHarmony 后这个链路必须原样跑通否则所有设备上报都会变成直连后端 HTTP 接口一旦流量抖动后端就会被打爆。这篇文章就是那次完整适配过程的复盘里面包含方案取舍、MethodChannel 桥接细节、SigV4 签名在 ArkTS 侧的实现以及我在生产环境里踩过的坑希望能给同样在 OpenHarmony 上做 Flutter 插件适配的同学省点时间。1. 这个库到底是干什么的分布式消息异步解耦的切入点1.1 为什么选 SQS 而不是其他消息中间件先聊聊背景。我们的场景是物联网设备端上报设备数量上千台每台每隔几秒就会产生一条状态数据。如果设备直连后端 API高峰期每秒可能有上千个并发请求后端服务要么疯狂扩容要么直接限流丢数据。用消息队列做中转以后设备端只负责把消息丢进队列后端按自己的处理能力去拉取两端互不阻塞这就是典型的异步解耦。技术选型时我们对比过 RabbitMQ、Kafka 和 AWS SQS。自建 RabbitMQ 或 Kafka 在云端要考虑运维成本而且我们的客户端是 Flutter 写的需要找 Dart 生态里维护活跃的 SDK。AWS SQS 虽然是云厂商托管服务但胜在完全不用运维标准队列无限吞吐还有死信队列、延迟队列、长轮询这些开箱即用的能力。配合aws_sqs_api这个纯 Dart 包Dart 层直接调用 SQS 的 REST API省掉了中间再套一层自建网关的成本。1.2 aws_sqs_api 的功能边界与依赖关系aws_sqs_api是 AWS 官方为 Dart 语言生成的 SQS API 客户端底层用的是 AWS 的 Smithy 代码生成框架。它本身不包含 UI 组件也不依赖任何 Flutter 原生插件核心能力就是封装 SQS 的 REST API 调用包括创建队列、发送消息、接收消息、删除消息、修改可见性超时等。它有几个关键依赖包需要一起引入aws_common提供 AWS 服务的通用基础类型和配置aws_signature_v4实现 AWS Signature Version 4 请求签名aws_smithy_clientSmithy 客户端运行时负责 HTTP 请求的发送和响应解析这个依赖关系很重要。aws_signature_v4是纯 Dart 实现的签名算法理论上在任何能跑 Dart 的平台上都能运行。但实际适配 OpenHarmony 时问题往往出在更底层——Dart 运行时能不能正常发 HTTPS 请求、网络权限怎么配、凭证存哪里。搞清楚这些边界你就知道鸿蒙适配的重点其实不在 Dart 层而在平台桥接层。2. 鸿蒙适配的核心难点拆解2.1 三层问题运行时、签名、原生通道把aws_sqs_api搬到 OpenHarmony我把它拆成了三个层次的问题逐个击破第一层是 Dart 运行时兼容性。OpenHarmony 上的 Flutter 是基于 OpenHarmony 官方移植的 Flutter SDK 来跑的大部分dart:io的能力都支持但跟 Android/iOS 的 Flutter 运行时不是同一个实现。我们在适配过程中发现aws_smithy_client里的某些网络异常处理在 OpenHarmony 上的表现略有差异具体来说是SocketException的报错信息格式不一样导致日志解析逻辑需要微调。这一层的问题比较隐蔽建议适配时先写一个最小化的 Dart 脚本在目标设备上跑一遍确认 HttpClient 能正常访问外网。第二层是 SigV4 签名算法的平台差异。aws_signature_v4用的是纯 Dart 的crypto包做 SHA256 和 HMAC 计算在 OpenHarmony 的 Dart 运行时上可以正常工作。但这里有个坑SQS 的请求签名要求 CanonicalRequest 里的 host 头必须和实际请求的 host 完全一致包括大小写和端口号。在 OpenHarmony 上如果你走了代理或者自定义了网络栈host 头可能会被改写导致服务端返回SignatureDoesNotMatch。第三层是原生平台通道。项目里的凭证信息之前是存在系统钥匙串里的Android 用的是flutter_secure_storageiOS 用 Keychain。OpenHarmony 上没有现成的插件这层必须自己写原生桥接。另外我们的业务还要求 App 在后台时也能持续消费队列消息这需要鸿蒙端的任务后台执行能力配合不是一个纯 Dart 包能解决的。所以适配工作的重心最终落在了 MethodChannel 的建联和 ArkTS 原生侧的实现上。2.2 MethodChannel 桥接 vs 纯 Dart 直连的取舍有一种思路是既然aws_sqs_api是纯 Dart 包OpenHarmony 的 Flutter 运行时又支持dart:io那是不是什么都不用改直接跑就完事了我最初也是这么想的在开发机上跑了个 demo还真能通。但放到生产环境就暴露了三个问题凭证存储没有安全的地方。Dart 侧只能用 shared_preferences 之类的插件存明文这在合规审计上过不去。后台消费不可靠。Flutter 的 Dart isolate 在应用退到后台后可能被系统挂起没有鸿蒙端原生任务配合消息消费会断。网络栈不可控。某些定制 ROM 的 OpenHarmony 设备会对 Flutter 的 HttpClient 做限制而走系统ohos.net.http是经过充分验证的通道。所以最终方案是Dart 层通过 MethodChannel 调 ArkTS 原生实现把发送消息、接收消息、删除消息、修改可见性这几个核心操作全部下沉到鸿蒙侧。aws_sqs_api在 Dart 层保留作为接口定义和数据模型参考真正发 HTTP 请求的是 ArkTS 代码。这个方案虽然多写了不少原生代码但换来的是安全存储、稳定网络和后台执行能力我认为是值得的。3. 实操从创建插件工程到跑通第一条消息3.1 工程结构配置与权限声明OpenHarmony 的 Flutter 插件和 Android 插件结构很像但目录名从android换成了ohos。我用的是手动创建的方式因为flutter create --templateplugin默认不支持生成 ohos 目录。工程目录结构长这样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里要声明插件支持的平台注意要加上 ohosflutter: plugin: platforms: android: package: com.example.aws_sqs_api_ohos pluginClass: AwsSqsApiPlugin ios: pluginClass: AwsSqsApiPlugin ohos: pluginClass: AwsSqsApiPlugin pluginImplementation: AwsSqsApiPluginImplmodule.json5里必须声明网络权限这是最容易漏的一步。鸿蒙应用默认没有网络访问权限不加这个权限所有 HTTPS 请求都会静默失败{ module: { name: entry, requestPermissions: [ { name: ohos.permission.INTERNET } ] } }这个权限配置和 Android 的AndroidManifest.xml里加uses-permission android:nameandroid.permission.INTERNET /是同一个作用但位置完全不同很多从 Android 转过来的同学会下意识去找 manifest 文件结果在鸿蒙工程里根本找不到。3.2 Dart 侧封装MethodChannel 的调用契约Dart 侧的封装尽量保持和原来aws_sqs_api的调用风格一致这样业务代码不用大面积改动。我定义了一个统一的方法通道名aws_sqs_api然后按 SQS 的核心操作拆成几个方法。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, }); } }注意几个设计细节receiveMessage的返回值我用了ListMapString, dynamic而不是强类型对象因为 MethodChannel 的 JSON 反序列化在鸿蒙端的Map键值类型可能和 Dart 侧不完全匹配留一层动态类型可以减少类型转换异常。所有方法名都用了小写驼峰因为 ArkTS 侧解析 MethodCall 时方法名是直接字符串匹配风格统一能减少低级错误。invokeMethod内部可以传MapString, Object?但嵌套 map 的 value 类型在跨通道传输时会被序列化成 JSON所以messageAttributes这里我限制成MapString, String避免复杂结构中int和double在 JSON 解析时的边界问题。3.3 ArkTS 侧实现SigV4 签名与 HTTPS 请求ArkTS 侧的插件实现是整个适配的核心。首先要实现 FlutterPlugin 接口在onAttachToFlutterEngine里注册 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类里实现具体的 SQS API 调用。这里最绕的是 SigV4 签名我把它拆成了几个工具方法。先看核心的签名逻辑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); } }这里我对每一步展开说明。buildPayload会把 SQS 的请求参数拼成application/x-www-form-urlencoded格式这是 SQS REST API 的标准格式。实际的请求体长这样ActionSendMessageVersion2012-11-05QueueUrlhttps%3A%2F%2Fsqs.us-east-1.amazonaws.com%2F123456789012%2Fmy-queueMessageBodyhelloSigV4 签名的计算我用的是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 }; }写这部分的时候我踩了一个很深的坑cryptoFramework的hmacSha256返回的是Uint8Array直接转字符串会拿到乱码必须先把 key 转成Uint8Array再做二进制拼接。上面代码里hmacSha256(kDate, requestInfo.region)这里的kDate是上一轮的二进制输出不能直接toString()否则签名结果永远和服务端对不上。4. 消费者侧的高可用设计4.1 可见性超时与消费失败处理消息发得出去不算完消费端的高可用才是真正考验设计功底的地方。SQS 的消息模型是拉取后隐藏消费者调用ReceiveMessage拿到消息后这条消息并不会立刻从队列删除而是进入不可见状态。这个不可见时间就叫 Visibility Timeout可见性超时。理解这个机制非常重要。如果消费者在超时时间内没有调用DeleteMessage删除消息SQS 会认为消费失败把消息重新放回队列再次对消费者可见。这就像你从快递柜取了个包裹但没在时限内拿走柜门会重新打开包裹又变成待取状态。OpenHarmony 客户端上我设置的默认可见性超时是 30 秒但实际业务处理完一条消息的平均耗时只有 2 到 3 秒。为什么留这么大的余量因为设备端的网络状况不稳定弱网环境下 SQS 的响应可能会延迟如果超时设得太短很容易造成消息在业务还没处理完时就被重新推送导致重复消费。如果超时设得太长又要担心消费者崩溃后消息长时间无人处理。我的处理策略是拉取到消息后立刻调用一次ChangeMessageVisibility把超时时间调整到 60 秒给业务处理预留充足时间业务处理成功后调用DeleteMessage删除消息。如果业务处理失败不调用删除让消息在超时后自动回到队列实现天然的重试机制。4.2 长轮询与批量拉取SQS 的消费者如果频繁轮询空队列会产生大量无效 API 调用既费钱又费电。Wi-Fi 环境下这个问题不明显但 OpenHarmony 设备往往是带电池的功耗控制很关键。SQS 提供了长轮询机制在ReceiveMessage请求里带WaitTimeSeconds参数可以设置 1 到 20 秒。当队列为空时请求不会立刻返回空列表而是挂住等待新消息到来或者直到超时时间结束。这样消费者每 20 秒只需要发起一次请求功耗大幅下降。批量拉取方面SQS 限制单次ReceiveMessage最多返回 10 条消息。我在 ArkTS 侧做了循环拉取一次业务触发最多拉取 50 条分 5 个批次并行处理每批之间加一个 100ms 的间隔避免瞬间打满网络带宽。实测下来在 2000 条消息积压的情况下消费完所有消息只需要 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 死信队列兜底再稳的系统也有处理不了的消息。比如设备上报了一条格式损坏的 JSON消费程序每次解析都会失败重试 10 次还是失败。如果任由这种消息在队列里反复横跳不仅浪费处理能力还会挤占正常消息的位置。SQS 的死信队列DLQ就是干这个的。在主队列的 Attributes 里配置 RedrivePolicy指定maxReceiveCount为 3 或 5这样一条消息被拉取超过指定次数后SQS 会自动把它转移到对应的死信队列。死信队列里的消息可以等开发人员修复 bug 后重新投递回主队列或者直接人工处理。在 OpenHarmony 客户端的适配里我把死信队列的消费单独做了一个通道。正常情况下客户端只消费主队列死信队列的消费由后端来处理。客户端发现消息拉取次数异常时会记录日志并上报一条告警事件方便运维人员及时发现。5. 实测中的坑与排查技巧5.1 SignatureDoesNotMatch我排查了一天的签名问题这个错误绝对是我这次适配里耗时最长的问题。现象很简单在 Android 上跑得好好的同样的参数搬到 OpenHarmony 上就报SignatureDoesNotMatch: The request signature we calculated does not match the signature you provided. Check your AWS Secret Access Key and signing method.我第一反应是凭证问题反复检查了 AccessKey 和 SecretKey确认没问题。然后又怀疑是 ArkTS 的 HMAC 实现有 bug打印出签名值逐字节比对发现也没有问题。最后查到问题出在 CanonicalRequest 里的 host 头。Dart 的aws_signature_v4在签名时用的是小写 host比如sqs.us-east-1.amazonaws.com。但 ArkTS 的ohos.net.http在发送请求时某些版本会在 header 里自动加上一个默认的Host头而且这个 Host 头的格式可能是SQS.US-EAST-1.AMAZONAWS.COM全大写。SQS 服务端在验证签名时是区分大小写的host 头不一致签名自然对不上。解决方案是手动设置请求的 header把 host 头固定成小写。还有一次是为了兼容签名区域问题改配置也排查了很久最后统一用us-east-1测试环境验证才定位到是区域参数传递错误。这些都是第一线实操才会遇到的事。header: { Content-Type: application/x-www-form-urlencoded, Host: host.toLowerCase(), X-Amz-Date: signature.amzDate, Authorization: signature.authorization, }排查建议先在电脑上用 curl 模拟完整的 SQS 请求把签名过程中每一步的中间值打印出来再用同样的参数在 OpenHarmony 设备上跑对比两个中间值哪里开始不一致。这个方法我屡试不爽。5.2 消息积压消费者线程被系统挂起了OpenHarmony 对后台任务的限制比 Android 更严格。应用退到后台后如果没有任何前台服务或长时任务在运行ArkTS 侧执行网络请求的协程会在几分钟内被系统挂起。表现就是应用在后台时消息不消费回到前台后突然开始大量消费积压消息。解决思路有两个方向我最终都做了在模块的module.json5里声明长时任务权限参考常见鸿蒙适配方案申请后台任务类型并配置对应的权限这样应用在后台运行时有系统级别的资源保障。在 ArkTS 侧用 WorkSchedulerExtension 定期唤醒每次唤醒拉取一批消息处理完再让系统休眠。实测下来消息积压时间窗口从原来的 10 分钟以上控制到了 1 分钟以内。5.3 重复消费正确使用 ReceiptHandleSQS 的消费模型是 at-least-once也就是至少一次不保证恰好一次。重复消费的根源在于网络超时比如客户端已经调用了DeleteMessage但响应在传输过程中丢失服务端没收到删除指令超时后消息再次变得可见。要减少重复消费唯一可靠的手段是让消费逻辑幂等。我在设备端对每条消息计算了一个业务唯一 ID写进MessageAttributes的messageId字段。消费端在处理前先查一下本地数据库如果这个 ID 已经处理过直接跳过。这个方案不能说 100% 杜绝重复但能把影响降到可以忽略的程度。5.4 常见问题速查表问题现象可能原因排查方向SignatureDoesNotMatchhost 头大小写不一致检查请求 header 中的 Host 是否为小写AccessDenied凭证错误或区域不匹配检查 AccessKey/SecretKey确认 region 参数QueueDoesNotExistQueueUrl 填错或权限不足检查 QueueUrl 的完整路径确认队列和凭证归属同一账号MethodChannel 调用超时ArkTS 侧网络请求阻塞检查网络权限确认module.json5中已声明 INTERNET消息积压且应用在后台后台任务被挂起配置长时任务权限或使用 WorkSchedulerExtension后台拿不到动态凭证凭证刷新逻辑没跑后台在 ArkTS 侧启动定时刷新保证 sessionToken 不过期5.5 凭证管理的安全实践最后单独聊聊凭证。AWS 的凭证如果写死到客户端里逆向出一个就能刷爆你的队列。我在 ArkTS 侧做了一层封装凭证不会明文存储在本地用系统的凭据加密能力加密后存入应用沙箱。每次 Build 时从服务端拉取临时凭证搭配 STS 的 sessionToken 使用过期后自动刷新。这样即使设备被 root泄露的也只是一段时间内的临时凭证影响范围可控。6. 这套方案的后续扩展方向把aws_sqs_api在 OpenHarmony 上跑通不是终点它给后续的架构演进留了好几个口子。一个是消息类型的扩展。现在发送的消息体是普通字符串但 SQS 的MessageAttributes支持结构化属性可以在发送时打上设备类型、环境、业务标签消费端根据这些属性做路由和处理策略分流。另一个是队列策略的调整。SQS 有个延时队列功能可以把消息延迟 0 到 900 秒后再对消费者可见。这个能力可以用来做设备升级的时间窗口控制比如设备收到升级指令后不用立刻执行而是先把指令投递到延迟队列过 15 分钟再拉取执行避开业务高峰。最后说一句实在话OpenHarmony 的 Flutter 生态还在快速完善阶段很多在三方库上的适配工作没有太多现成资料可查。遇到问题多看官方文档、多打印日志、多跟同类项目的开发者交流比自己闷头排查高效得多。这篇复盘里写的坑都是我实实在在踩过的能帮你少走几步弯路就是它最大的价值。
