HarmonyOS WebSocket：发送完成之后，还差一份业务回执【共创稿事节】
连接建立、发送成功、业务完成，早期页面经常把这三件事挤进同一条绿色提示。改成持续连接以后，这种写法更容易暴露问题：发送接口已经返回，服务端却可能还没有完成自己的检查；连接随后断开，应用甚至无法知道对方究竟处理到了哪一步。页面需要一份由业务协议定义的回执，才能把这些含义分开。
从原样回声升级为有含义的回复
EchoWire 工程只做一件事：连接本地测试服务，发送固定测试文本，等待同编号的接受回执。它不接入聊天室、不自动重发、不申请后台持续运行能力。范围保持集中，方便观察一条消息从提交到得到确认的完整过程，而不是用大量界面功能掩盖通信状态。
工程使用 API 12、Stage 模型，通过 Network Kit 创建连接。官方接口参考要求先订阅连接打开事件，再发起连接，并在连接打开后发送。连接方法的返回结果不能代替所有后续事件，页面也不能在用户刚点连接按钮时就立即允许发送。
旧的回声服务只把收到的字符串原样发回，可以证明两端存在消息往返，却没有表达服务端是否接受了某项业务。本例把请求定义为版本、类型、编号和文本四个字段；回复也带版本、类型和相同编号，并明确标记接受。这个标记仅表示下方测试服务完成格式校验，不代表文件落盘或其他未实现动作。
请求编号在当前页面内递增，同时只允许一条等待中的请求。这样既能识别不属于当前请求的消息，也能把界面状态保持简单。业务增长后可以扩展为等待表，但在没有真正需要并发之前，一开始就加入复杂队列会增加验证范围，也让回执与按钮状态更难对齐。
新建 common/WireProtocol.ets。类型断言只帮助编译器理解字段，不会检查网络数据，所以解析后仍逐项验证。本文只接受短文本回执，二进制消息直接拒绝，不把字节缓冲随意转换成字符串后继续猜测。
// entry/src/main/ets/common/WireProtocol.ets
export interface WireAck { v: number; kind: string; id: number; accepted: boolean; }
export interface WireRequest { v: number; kind: string; id: number; text: string; }
export function readAck(raw: string | ArrayBuffer): WireAck {
if (typeof raw !== 'string' || raw.length > 512) {
throw new Error('只接受短文本回执');
}
const ack = JSON.parse(raw) as WireAck;
if (ack === null || typeof ack !== 'object' || Array.isArray(ack) ||
ack.v !== 1 || ack.kind !== 'ack' || ack.accepted !== true ||
!Number.isSafeInteger(ack.id) || ack.id < 1) {
throw new Error('回执字段不符合协议');
}
return ack;
}
编号不能单独承担所有身份含义。本例使用连接对象隔离会话，并且服务不在连接之间保存历史回执，因此页面内递增编号足以关联当前示例消息。如果要跨应用重启恢复业务，就需要持久、可去重的业务标识和服务端结果记录，不能把一个从一重新计数的变量当成全局唯一凭证。
回执协议也不等于底层心跳。心跳可以帮助判断链路是否仍有响应，却不能证明上一条业务已经完成检查。把心跳包显示成业务成功，会产生比没有状态更难排查的误导；客户端应按消息类型分别处理，每一种类型都有自己的字段和状态更新规则。
准备一个能独立复现的本地服务
在开发电脑新建空目录，安装 Node.js 与 npm 后执行下面的命令。这里使用 ws 的第八主版本；安装得到的锁文件应随本地测试目录保留，便于团队记录实际依赖版本。项目官方示例提供了服务器与消息事件的用法，下面加入的是本工程自己的校验和回执约定。
npm init -y
npm install ws@8
node server.mjs
先创建以下 server.mjs，再执行最后一条启动命令。服务监听开发电脑的本地网络接口，设备必须能访问该电脑的对应地址与端口。只在可信局域网做这个无鉴权测试，不开放到公网；正式服务应另行加入身份校验与受信任证书下的加密连接。
// server.mjs：运行在开发电脑，不放进 ArkTS 页面目录
import { WebSocketServer } from 'ws';
const server = new WebSocketServer({
host: '0.0.0.0', port: 8765, maxPayload: 1024, perMessageDeflate: false
});
server.on('error', (error) => { console.error('监听失败：', error.message); });
server.on('listening', () => { console.log('本地测试服务监听 8765'); });
server.on('connection', (socket) => {
socket.on('error', (error) => { console.error('连接错误：', error.message); });
socket.on('message', (data, isBinary) => {
if (isBinary) { socket.close(1003, 'text only'); return; }
try {
const request = JSON.parse(data.toString('utf8'));
if (request === null || typeof request !== 'object' || Array.isArray(request) ||
request.v !== 1 || request.kind !== 'probe' ||
!Number.isSafeInteger(request.id) || request.id < 1 ||
typeof request.text !== 'string' || request.text.length < 1 ||
request.text.length > 64) {
socket.close(1008, 'bad request'); return;
}
const ack = { v: 1, kind: 'ack', id: request.id, accepted: true };
socket.send(JSON.stringify(ack), (error) => {
if (error) { console.error('回执发送失败：', error.message); }
});
} catch {
socket.close(1008, 'invalid json');
}
});
});
服务端限制的是消息字节大小，业务字段另有限制字符长度，两者作用不同。短文本并不保证字节数同样短，尤其包含中文时更需要分清。客户端只发固定短样本，因此不会接近这个上限；这些参数是本例的协议约束，不是网络库建议的通用容量。
在 entry/src/main/module.json5 的 module 对象里加入网络权限，保留模板的其他配置。该权限由系统授予，不需要再弹出运行时申请框。设备地址框填写开发电脑实际可达的地址，例如使用“ws://电脑的局域网地址:8765”这一结构；手机上的本机回环地址指向手机自身，通常不能指向开发电脑。
"requestPermissions": [
{ "name": "ohos.permission.INTERNET" }
]
启动服务之后先检查电脑是否确实监听成功，再确认设备与电脑之间的网络是否允许互访。若地址不可达，应该先排查地址、端口和网络隔离，再研究回执字段。连接尚未打开时，业务解析函数甚至没有运行机会，把网络配置问题当成消息结构错误会浪费排查时间。
服务代码并不保存请求，也不执行有副作用的业务。它回复接受，只能支持“这份请求通过了本地规则”这一结论。若后续接入真实写操作，应把回执放在约定的业务阶段，并在协议说明中写明是已入队、已提交还是已持久化，不能沿用一个模糊的成功词覆盖多个阶段。
连接事件与业务回执各更新自己的状态
页面使用普通字段持有连接对象与请求编号，界面字段只描述用户需要看到的状态。所有事件都检查当前对象身份，旧连接不会修改新连接的页面。释放函数先撤销当前引用，再取消订阅、清理定时器和关闭连接，使后到达的事件失去写入资格。
// entry/src/main/ets/pages/Index.ets
import { webSocket } from '@kit.NetworkKit';
import { BusinessError } from '@kit.BasicServicesKit';
import { readAck, WireAck, WireRequest } from '../common/WireProtocol';
@Entry
@Component
struct Index {
@State endpoint: string = '';
@State phase: string = '未连接';
@State receipt: string = '尚无回执';
@State waiting: boolean = false;
private socket: webSocket.WebSocket | undefined = undefined;
private pendingId: number = 0;
private serial: number = 0;
private timer: number = -1;
private alive: boolean = true;
aboutToAppear(): void { this.alive = true; }
aboutToDisappear(): void { this.alive = false; this.release(); }
private current(ws: webSocket.WebSocket): boolean {
return this.alive && this.socket === ws;
}
private clearTimer(): void {
if (this.timer >= 0) { clearTimeout(this.timer); this.timer = -1; }
}
private release(): void {
const ws = this.socket;
this.socket = undefined;
this.pendingId = 0;
this.clearTimer();
if (ws === undefined) { return; }
try {
ws.off('open'); ws.off('message'); ws.off('close'); ws.off('error');
ws.close().catch((error: BusinessError) => {
console.warn(`关闭连接失败：${error.code}`);
});
} catch (error) {
console.warn(`释放连接失败：${(error as BusinessError).code}`);
}
}
private stop(note: string): void {
this.release();
if (!this.alive) { return; }
this.phase = '未连接'; this.waiting = false; this.receipt = note;
}
private async connect(): Promise<void> {
if (this.socket !== undefined) { return; }
const url = this.endpoint.trim();
if (url.length > 1024 || !/^wss?:\/\/[^\s]+$/.test(url)) {
this.receipt = '请填写可达的 ws 或 wss 地址'; return;
}
let ws: webSocket.WebSocket;
try { ws = webSocket.createWebSocket(); } catch (error) {
this.receipt = `创建连接失败：${(error as BusinessError).code}`; return;
}
this.socket = ws;
this.phase = '连接中'; this.receipt = '等待连接打开';
try {
ws.on('open', (error: BusinessError, value: Object) => {
if (!this.current(ws)) { return; }
if (error) { this.stop(`连接失败：${error.code}`); return; }
this.phase = '已连接'; this.receipt = '可以发送测试消息';
});
ws.on('message', (error: BusinessError, raw: string | ArrayBuffer) => {
if (!this.current(ws)) { return; }
if (error) { this.stop(`接收失败：${error.code}`); return; }
try {
const ack: WireAck = readAck(raw);
if (ack.id !== this.pendingId || this.pendingId === 0) { return; }
this.clearTimer(); this.pendingId = 0; this.waiting = false;
this.receipt = `服务已接受消息 ${ack.id}`;
} catch (error) {
this.stop(`回执无效：${(error as Error).message}`);
}
});
ws.on('close', (error: BusinessError, result: webSocket.CloseResult) => {
if (!this.current(ws)) { return; }
this.stop(this.pendingId > 0 ? '连接已关闭，待确认消息结果未知' :
(error ? `关闭异常：${error.code}` : `连接已关闭：${result.code}`));
});
ws.on('error', (error: BusinessError) => {
if (this.current(ws)) { this.stop(`连接异常：${error.code}，未获回执不代表未处理`); }
});
const accepted = await ws.connect(url);
if (this.current(ws) && !accepted) { this.stop('连接请求未成功'); }
} catch (error) {
if (this.current(ws)) { this.stop(`连接失败：${(error as BusinessError).code}`); }
}
}
private async sendProbe(): Promise<void> {
const ws = this.socket;
if (ws === undefined || this.phase !== '已连接' || this.waiting) { return; }
if (this.serial >= Number.MAX_SAFE_INTEGER) { this.stop('编号耗尽，请重新打开页面'); return; }
const id = ++this.serial;
const request: WireRequest = { v: 1, kind: 'probe', id: id, text: '连线测试' };
this.pendingId = id; this.waiting = true;
this.receipt = `消息 ${id} 等待业务回执`;
this.timer = setTimeout(() => {
if (this.current(ws) && this.pendingId === id) {
this.stop('回执等待结束，处理结果未知');
}
}, 8000);
try {
const sent = await ws.send(JSON.stringify(request));
if (this.current(ws) && this.pendingId === id && !sent) {
this.stop('发送未成功，处理结果仍需确认');
}
} catch (error) {
if (this.current(ws) && this.pendingId === id) {
this.stop(`发送异常：${(error as BusinessError).code}，处理结果未知`);
}
}
}
build() {
Column({ space: 16 }) {
Text('回执连线').fontSize(28).fontWeight(FontWeight.Bold)
Text('连接打开后，再确认业务接受').fontSize(14).fontColor('#64748B')
TextInput({ text: this.endpoint, placeholder: '填写测试服务地址' })
.height(52).fontSize(15).enabled(this.phase === '未连接')
.onChange((value: string) => { this.endpoint = value; })
Column({ space: 12 }) {
Text(this.phase).fontSize(26).fontWeight(FontWeight.Bold).fontColor('#0E7490')
Text(this.receipt).fontSize(16).textAlign(TextAlign.Center)
}.width('100%').padding(24).backgroundColor('#FFFFFF').borderRadius(20)
Button('连接服务').width('100%').height(44).backgroundColor('#0E7490')
.enabled(this.phase === '未连接').onClick(() => { this.connect(); })
Button('发送测试消息').width('100%').height(44).backgroundColor('#0E7490')
.enabled(this.phase === '已连接' && !this.waiting).onClick(() => { this.sendProbe(); })
Button('断开连接').width('100%').height(44).backgroundColor('#64748B')
.enabled(this.phase !== '未连接').onClick(() => {
this.stop(this.waiting ? '已主动断开，处理结果未知' : '已主动断开');
})
Text('发送成功 ≠ 业务已接受').fontSize(14).fontColor('#64748B')
}.width('100%').height('100%').padding(24).backgroundColor('#F0FAFC')
}
}
回执可能比发送方法的异步续段更早到达，所以发送续段只能处理仍在等待的同一编号，不能把已经接受的状态改回等待。这里没有在发送返回真时标记成功，真正的确认只发生在接收消息并核对协议之后。关闭和异常则终止本次会话，保留结果未知这一事实。
等待中的按钮被禁用，是为了表达本例只支持一条在途消息，而不是声称网络库只能发送一条。这个约束让超时、断开和回执都能围绕一个编号判断。如果以后允许多条请求并行，每条请求都需要独立的等待期限和最终状态，不能继续共用一个布尔字段。
地址输入也随连接阶段锁定。这样页面展示的地址始终对应当前尝试连接的目标，用户不会在连接尚未完成时把文本改成另一个地址，随后误以为结果来自新目标。重新连接前必须结束旧会话，新的地址才会被读取并创建新对象，界面与资源的关系因此保持清楚。
八秒是本地示例的等待预算，不是服务处理性能承诺。等待结束后主动关闭连接，阻止同一会话继续堆积不确定消息。它并不能证明服务没有处理请求；正式系统若涉及订单、写入或扣减，必须用可查询结果和幂等键解决重试，不能因为页面超时就盲目再发一次。
验证的是协议含义，不只是一盏连接灯
先启动服务，再填入设备可达的地址。连接打开前发送按钮不可用，打开后发送固定文本，应收到同编号回执并展示服务已接受。
随后分别修改测试服务的回执版本、类型和编号。版本或结构错误应关闭会话并报告回执无效；一个结构合法但编号不匹配的回执不会确认当前请求，最终进入等待结束。两种结果不同，前者表示协议破坏，后者表示消息不能归属于当前等待项。
还可以临时跳过服务器的发送回执语句，观察等待结束路径。再在消息发出后停止服务，检查页面是否错误显示“业务失败”或“已完成”。对这个最小协议来说，正确表述是没有获得确认，处理结果未知。测试要保留修改点，完成后恢复服务代码，避免下一次验证仍使用故意破坏的版本。
协议校验还应单独覆盖空字符串、普通文本、数组、空对象以及超长文本。它们都可能经过网络正常到达，但不是本应用接受的回执。测试的关注点是没有任何一种非法输入能把页面变成已接受，而不是要求所有解析错误都具有完全相同的异常信息。
一个已经确认过的编号再次到达时，等待编号已经归零，因此重复回执不会再推动状态。这个行为适合本例的单次确认，但不等于服务端请求已经实现去重。客户端忽略重复回复，与服务端避免重复执行是两个独立责任，前者无法代替后者。
主动断开与页面销毁也要分开看。前者更新用户可见的状态，后者只释放普通资源，不在销毁阶段写响应式字段。连接属于当前页面，页面销毁时结束；如果产品需要跨页面共享，应把所有权提升到独立服务，并重新定义订阅者数量和停止时机，而不是让页面对象长期留在回调里。
排障记录可以保留会话建立时间、请求编号、消息类型以及状态转换原因，不必输出完整业务正文。对这个工程，固定测试文本公开且无敏感内容；迁入真实业务后，沿用随手打印整份消息的习惯会扩大信息暴露范围。先保留足以关联问题的元信息，再按明确需要增加临时诊断字段。
协议版本的价值在于让变化可以被识别。将来接受回执需要增加业务结果码，应明确哪些版本可互通、哪些字段必须存在，以及旧客户端应该如何拒绝新结构。静默忽略关键字段看似兼容，却可能让用户在服务端未接受时看到成功；显式拒绝更容易发现双方版本不一致。
WebSocket 提供持续的双向消息通道，业务仍要定义消息代表什么。这个工程把连接打开、提交发送、收到接受回执和结果未知放在各自的位置。后续增加批量并发、断线恢复或结果查询时，也应沿着这条边界扩展，让每一条成功提示都能对应一个确切发生的事实。
参考资料
讨论HarmonyOS开发技术，专注于API与组件、DevEco Studio、测试、元服务和应用上架分发等。
更多推荐
0
0- 0
已为社区贡献40条内容
所有评论(0)