8 第7章 端云协同开发
8.1 7.1 端云协同架构概述
8.1.1 7.1.1 端云协同的设计理念
端云协同开发是HarmonyOS应用开发中的核心能力之一。所谓”端云协同”，是指将终端设备（手机、平板、穿戴设备等）的计算能力与云端服务器的计算能力有机结合，通过高效的通信机制和协同策略，实现应用功能的最佳分配与数据的高效流转。
端云协同的核心设计理念包括：
- 以端侧体验为中心：云端能力服务于端侧用户体验，确保低延迟、高可靠的交互响应。
- 能力互补：端侧擅长实时交互、传感器数据采集和本地缓存；云侧擅长海量存储、复杂计算和多端数据汇聚。
- 弹性伸缩：应用逻辑可根据网络状况、设备性能等因素，在端侧和云侧之间灵活调度。
- 数据一致性：通过同步机制保证端云数据的最终一致性，支持离线场景下的数据完整性。
8.1.2 7.1.2 架构模式
在HarmonyOS应用开发中，常见的端云协同架构模式有以下几种：
B/S架构（Browser/Server）
B/S架构适用于轻量级、内容展示型应用。端侧通过Web组件加载云端页面，业务逻辑主要在服务器端执行。
// Web组件加载云端页面
@Component
struct WebPage {
controller: WebviewController = new WebviewController()
build() {
Column() {
Web({ src: 'https://api.example.com/app', controller: this.controller })
.javaScriptAccess(true)
.domStorageAccess(true)
.onPageEnd(() => {
console.info('页面加载完成')
})
.onErrorReceive((event) => {
console.error(`加载错误: ${event?.result?.getErrorInfo()}`)
})
}
}
}C/S架构（Client/Server）
C/S架构适用于需要丰富本地交互的应用。端侧承担UI渲染和部分业务逻辑，云端提供数据API服务。
// C/S架构中的API服务调用
class ApiService {
private baseUrl: string = 'https://api.example.com/v1'
async getUserProfile(userId: string): Promise<UserProfile> {
const httpRequest = http.createHttp()
try {
const response = await httpRequest.request(
`${this.baseUrl}/users/${userId}/profile`,
{
method: http.RequestMethod.GET,
header: { 'Content-Type': 'application/json' },
connectTimeout: 10000,
readTimeout: 30000
}
)
if (response.responseCode === http.ResponseCode.OK) {
return JSON.parse(response.result as string) as UserProfile
}
throw new Error(`请求失败: ${response.responseCode}`)
} finally {
httpRequest.destroy()
}
}
}混合架构
混合架构结合了B/S和C/S的优势，是HarmonyOS应用中最常用的架构模式。核心功能使用原生组件实现，部分功能通过Web组件或云函数完成。
8.1.3 7.1.3 端云职责划分原则
合理的端云职责划分是端云协同开发的关键。以下是职责划分的基本原则：
| 职责维度 | 端侧承担 | 云侧承担 |
|---|---|---|
| UI渲染 | 原生组件渲染、动画执行 | 页面模板下发、动态配置 |
| 业务逻辑 | 实时交互逻辑、本地校验 | 复杂计算、规则引擎、风控 |
| 数据存储 | 本地缓存、Preferences、RDB | 持久化存储、数据备份 |
| 网络通信 | 请求发起、连接管理 | API网关、负载均衡 |
| 安全 | 本地加密、生物认证 | 证书管理、安全审计 |
8.1.4 7.1.4 通信协议选择
HarmonyOS端云通信支持多种协议，开发者需根据场景选择合适的协议：
HTTP/HTTPS：适用于标准的请求-响应模式，是RESTful API的基础协议。
import { http } from '@kit.NetworkKit'
class HttpApiClient {
private static instance: HttpApiClient | null = null
private baseUrl: string = 'https://api.example.com'
static getInstance(): HttpApiClient {
if (!HttpApiClient.instance) {
HttpApiClient.instance = new HttpApiClient()
}
return HttpApiClient.instance
}
async get<T>(path: string, params?: Record<string, string>): Promise<T> {
const httpRequest = http.createHttp()
try {
let url = `${this.baseUrl}${path}`
if (params) {
const queryString = Object.entries(params)
.map(([k, v]) => `${encodeURIComponent(k)}=${encodeURIComponent(v)}`)
.join('&')
url += `?${queryString}`
}
const response = await httpRequest.request(url, {
method: http.RequestMethod.GET,
header: {
'Content-Type': 'application/json',
'Authorization': `Bearer ${await this.getAccessToken()}`
},
connectTimeout: 15000,
readTimeout: 30000
})
if (response.responseCode === http.ResponseCode.OK) {
return JSON.parse(response.result as string) as T
}
throw new ApiError(response.responseCode, `HTTP错误: ${response.responseCode}`)
} finally {
httpRequest.destroy()
}
}
async post<T>(path: string, body: object): Promise<T> {
const httpRequest = http.createHttp()
try {
const response = await httpRequest.request(`${this.baseUrl}${path}`, {
method: http.RequestMethod.POST,
header: {
'Content-Type': 'application/json',
'Authorization': `Bearer ${await this.getAccessToken()}`
},
extraData: JSON.stringify(body),
connectTimeout: 15000,
readTimeout: 30000
})
if (response.responseCode === http.ResponseCode.OK ||
response.responseCode === http.ResponseCode.CREATED) {
return JSON.parse(response.result as string) as T
}
throw new ApiError(response.responseCode, `请求失败`)
} finally {
httpRequest.destroy()
}
}
private async getAccessToken(): Promise<string> {
// 从安全存储中获取Token
return 'mock_token'
}
}
class ApiError extends Error {
code: number
constructor(code: number, message: string) {
super(message)
this.code = code
}
}WebSocket：适用于需要实时双向通信的场景，如聊天、实时协作、行情推送等。
import { webSocket } from '@kit.NetworkKit'
class WebSocketClient {
private ws: webSocket.WebSocket | null = null
private reconnectTimer: number = -1
private maxReconnectAttempts: number = 5
private reconnectAttempts: number = 0
private reconnectInterval: number = 3000
private messageHandlers: Map<string, Array<(data: object) => void>> = new Map()
async connect(url: string): Promise<void> {
this.ws = webSocket.createWebSocket()
this.ws.on('open', (err: Error, value: Object) => {
console.info('WebSocket连接已建立')
this.reconnectAttempts = 0
this.startHeartbeat()
})
this.ws.on('message', (err: Error, value: string | ArrayBuffer) => {
this.handleMessage(value as string)
})
this.ws.on('close', (err: Error, value: webSocket.CloseResult) => {
console.info('WebSocket连接已关闭')
this.stopHeartbeat()
this.tryReconnect(url)
})
this.ws.on('error', (err: Error) => {
console.error(`WebSocket错误: ${err.message}`)
})
await this.ws.connect(url, {
header: {
'Authorization': 'Bearer token_here'
}
})
}
send(type: string, data: object): void {
if (this.ws) {
const message = JSON.stringify({ type, data, timestamp: Date.now() })
this.ws.send(message)
}
}
on(type: string, handler: (data: object) => void): void {
if (!this.messageHandlers.has(type)) {
this.messageHandlers.set(type, [])
}
this.messageHandlers.get(type)!.push(handler)
}
private handleMessage(rawMessage: string): void {
try {
const message = JSON.parse(rawMessage)
const handlers = this.messageHandlers.get(message.type)
if (handlers) {
handlers.forEach(handler => handler(message.data))
}
} catch (e) {
console.error('消息解析失败')
}
}
private heartbeatTimer: number = -1
private startHeartbeat(): void {
this.heartbeatTimer = setInterval(() => {
this.send('ping', {})
}, 30000) as number
}
private stopHeartbeat(): void {
if (this.heartbeatTimer !== -1) {
clearInterval(this.heartbeatTimer)
this.heartbeatTimer = -1
}
}
private tryReconnect(url: string): void {
if (this.reconnectAttempts < this.maxReconnectAttempts) {
this.reconnectAttempts++
console.info(`尝试重连 (${this.reconnectAttempts}/${this.maxReconnectAttempts})`)
this.reconnectTimer = setTimeout(() => {
this.connect(url)
}, this.reconnectInterval * this.reconnectAttempts) as number
}
}
disconnect(): void {
if (this.reconnectTimer !== -1) {
clearTimeout(this.reconnectTimer)
}
this.stopHeartbeat()
if (this.ws) {
this.ws.close()
this.ws = null
}
}
}gRPC：适用于高性能、低延迟的RPC调用场景，支持Protocol Buffers序列化，在微服务架构中广泛使用。HarmonyOS可通过HTTP/2结合Protobuf实现类似gRPC的通信能力。
8.2 7.2 网络请求管理
8.2.1 7.2.1 @ohos.net.http模块详解
@ohos.net.http（即@kit.NetworkKit中的http模块）是HarmonyOS提供的HTTP网络请求核心模块。它支持GET、POST、PUT、DELETE等标准HTTP方法，支持请求头自定义、超时控制、响应码处理等功能。
import { http } from '@kit.NetworkKit'
// 创建HTTP请求对象
const httpRequest = http.createHttp()
// 发起HTTP请求
async function fetchData(): Promise<void> {
try {
const response = await httpRequest.request('https://api.example.com/data', {
method: http.RequestMethod.POST,
header: {
'Content-Type': 'application/json',
'Accept': 'application/json'
},
extraData: JSON.stringify({
query: 'test',
page: 1,
pageSize: 20
}),
connectTimeout: 10000, // 连接超时10秒
readTimeout: 30000, // 读取超时30秒
usingCache: true, // 启用缓存
usingProtocol: http.HttpProtocol.HTTP1_1 // 使用HTTP/1.1
})
console.info(`响应码: ${response.responseCode}`)
console.info(`响应头: ${JSON.stringify(response.header)}`)
console.info(`响应体: ${response.result}`)
} catch (error) {
console.error(`请求异常: ${error.message}`)
} finally {
// 务必销毁请求对象释放资源
httpRequest.destroy()
}
}8.2.2 7.2.2 请求拦截器与响应拦截器
在实际项目中，通常需要对所有请求进行统一处理，如添加认证头、日志记录、错误统一处理等。通过实现拦截器机制，可以优雅地完成这些需求。
// 请求配置类型
interface RequestConfig {
url: string
method: http.RequestMethod
headers: Record<string, string>
body?: string
connectTimeout?: number
readTimeout?: number
}
// 响应结果类型
interface ResponseResult<T = Object> {
code: number
data: T
message: string
}
// 拦截器接口
interface RequestInterceptor {
intercept(config: RequestConfig): RequestConfig | Promise<RequestConfig>
}
interface ResponseInterceptor {
intercept(response: ResponseResult): ResponseResult | Promise<ResponseResult>
interceptError(error: Error): void | Promise<void>
}
// HTTP客户端封装
class HttpClient {
private requestInterceptors: RequestInterceptor[] = []
private responseInterceptors: ResponseInterceptor[] = []
private baseUrl: string = ''
setBaseUrl(url: string): void {
this.baseUrl = url
}
addRequestInterceptor(interceptor: RequestInterceptor): void {
this.requestInterceptors.push(interceptor)
}
addResponseInterceptor(interceptor: ResponseInterceptor): void {
this.responseInterceptors.push(interceptor)
}
async request<T>(config: RequestConfig): Promise<T> {
// 执行请求拦截器链
let processedConfig = config
for (const interceptor of this.requestInterceptors) {
processedConfig = await interceptor.intercept(processedConfig)
}
const httpRequest = http.createHttp()
try {
const response = await httpRequest.request(processedConfig.url, {
method: processedConfig.method,
header: processedConfig.headers,
extraData: processedConfig.body,
connectTimeout: processedConfig.connectTimeout ?? 15000,
readTimeout: processedConfig.readTimeout ?? 30000
})
let result: ResponseResult = JSON.parse(response.result as string)
// 执行响应拦截器链
for (const interceptor of this.responseInterceptors) {
result = await interceptor.intercept(result)
}
return result.data as T
} catch (error) {
// 执行错误拦截器
for (const interceptor of this.responseInterceptors) {
await interceptor.interceptError(error as Error)
}
throw error
} finally {
httpRequest.destroy()
}
}
}
// 认证拦截器实现
class AuthInterceptor implements RequestInterceptor {
async intercept(config: RequestConfig): Promise<RequestConfig> {
const token = await TokenManager.getInstance().getValidToken()
if (token) {
config.headers['Authorization'] = `Bearer ${token}`
}
return config
}
}
// 日志拦截器实现
class LogInterceptor implements RequestInterceptor, ResponseInterceptor {
intercept(config: RequestConfig): RequestConfig {
console.info(`[HTTP] ${config.method} ${config.url}`)
return config
}
intercept(response: ResponseResult): ResponseResult {
console.info(`[HTTP Response] code: ${response.code}, message: ${response.message}`)
return response
}
interceptError(error: Error): void {
console.error(`[HTTP Error] ${error.message}`)
}
}
// 使用示例
const client = new HttpClient()
client.setBaseUrl('https://api.example.com')
client.addRequestInterceptor(new AuthInterceptor())
client.addRequestInterceptor(new LogInterceptor() as RequestInterceptor)
client.addResponseInterceptor(new LogInterceptor())8.2.3 7.2.3 请求队列与并发控制
在高并发场景下，需要控制同时发起的网络请求数量，避免资源耗尽或服务端过载。
// 请求队列与并发控制器
class RequestQueue {
private maxConcurrency: number
private runningCount: number = 0
private queue: Array<() => Promise<void>> = []
constructor(maxConcurrency: number = 5) {
this.maxConcurrency = maxConcurrency
}
async enqueue<T>(task: () => Promise<T>): Promise<T> {
return new Promise<T>((resolve, reject) => {
const wrappedTask = async () => {
try {
this.runningCount++
const result = await task()
resolve(result)
} catch (error) {
reject(error)
} finally {
this.runningCount--
this.processNext()
}
}
if (this.runningCount < this.maxConcurrency) {
wrappedTask()
} else {
this.queue.push(wrappedTask)
}
})
}
private processNext(): void {
if (this.queue.length > 0 && this.runningCount < this.maxConcurrency) {
const nextTask = this.queue.shift()
if (nextTask) {
nextTask()
}
}
}
clear(): void {
this.queue = []
}
get pendingCount(): number {
return this.queue.length
}
get activeCount(): number {
return this.runningCount
}
}
// 使用示例
const requestQueue = new RequestQueue(3) // 最大并发3个请求
async function batchFetchData(ids: string[]): Promise<UserProfile[]> {
const tasks = ids.map(id =>
requestQueue.enqueue(() => fetchUserProfile(id))
)
return Promise.all(tasks)
}8.2.4 7.2.4 超时与重试策略
网络请求的超时和重试策略是保证应用可靠性的关键。
// 带重试策略的请求封装
class RetryableRequest {
private maxRetries: number
private baseDelay: number
private maxDelay: number
constructor(maxRetries: number = 3, baseDelay: number = 1000, maxDelay: number = 10000) {
this.maxRetries = maxRetries
this.baseDelay = baseDelay
this.maxDelay = maxDelay
}
async execute<T>(requestFn: () => Promise<T>): Promise<T> {
let lastError: Error | null = null
for (let attempt = 0; attempt <= this.maxRetries; attempt++) {
try {
return await requestFn()
} catch (error) {
lastError = error as Error
// 判断是否可重试（仅对网络错误和5xx错误重试）
if (!this.isRetryable(error as Error)) {
throw error
}
if (attempt < this.maxRetries) {
// 指数退避策略
const delay = Math.min(
this.baseDelay * Math.pow(2, attempt) + Math.random() * 1000,
this.maxDelay
)
console.info(`请求失败，${delay}ms后重试 (${attempt + 1}/${this.maxRetries})`)
await this.sleep(delay)
}
}
}
throw lastError!
}
private isRetryable(error: Error): boolean {
// 网络超时、连接错误等可重试
const retryableMessages = ['timeout', 'connection', 'network', 'ECONNRESET']
return retryableMessages.some(msg =>
error.message.toLowerCase().includes(msg)
)
}
private sleep(ms: number): Promise<void> {
return new Promise(resolve => setTimeout(resolve, ms))
}
}
// 使用示例
const retryable = new RetryableRequest(3, 1000, 8000)
async function fetchWithRetry(): Promise<UserProfile> {
return retryable.execute(() => fetchUserProfile('user_001'))
}8.2.5 7.2.5 网络状态监听与离线处理
HarmonyOS提供了网络状态监听API，应用可以实时感知网络变化并做出相应处理。
import { connection } from '@kit.NetworkKit'
// 网络状态管理器
class NetworkMonitor {
private static instance: NetworkMonitor | null = null
private netConnection: connection.NetConnection | null = null
private isOnline: boolean = true
private listeners: Array<(online: boolean) => void> = []
static getInstance(): NetworkMonitor {
if (!NetworkMonitor.instance) {
NetworkMonitor.instance = new NetworkMonitor()
}
return NetworkMonitor.instance
}
startMonitoring(): void {
this.netConnection = connection.createNetConnection()
// 网络可用
this.netConnection.on('netAvailable', (netHandle: connection.NetHandle) => {
console.info('网络已连接')
this.isOnline = true
this.notifyListeners(true)
})
// 网络丢失
this.netConnection.on('netLost', (netHandle: connection.NetHandle) => {
console.warn('网络已断开')
this.isOnline = false
this.notifyListeners(false)
})
// 网络能力变化
this.netConnection.on('netCapabilitiesChange',
(netHandle: connection.NetHandle, netCap: connection.NetCapabilityInfo) => {
console.info('网络能力变化')
})
this.netConnection.register((error) => {
if (error) {
console.error(`网络监听注册失败: ${error.message}`)
}
})
}
onNetworkChange(listener: (online: boolean) => void): void {
this.listeners.push(listener)
}
getOnlineStatus(): boolean {
return this.isOnline
}
private notifyListeners(online: boolean): void {
this.listeners.forEach(listener => listener(online))
}
stopMonitoring(): void {
if (this.netConnection) {
this.netConnection.unregister((error) => {
if (error) {
console.error(`取消网络监听失败: ${error.message}`)
}
})
}
}
}
// 离线数据缓存管理器
class OfflineCacheManager {
private pendingRequests: Array<{ url: string; config: object; timestamp: number }> = []
// 离线时缓存请求
cacheRequest(url: string, config: object): void {
this.pendingRequests.push({
url,
config,
timestamp: Date.now()
})
// 持久化到本地
this.saveToStorage()
}
// 网络恢复后同步缓存的请求
async syncPendingRequests(): Promise<void> {
const httpRequest = http.createHttp()
const failedRequests: typeof this.pendingRequests = []
for (const request of this.pendingRequests) {
try {
// 检查缓存是否过期（超过24小时的请求不再重试）
if (Date.now() - request.timestamp > 24 * 60 * 60 * 1000) {
continue
}
await httpRequest.request(request.url, request.config as http.HttpRequestOptions)
} catch (error) {
failedRequests.push(request)
}
}
this.pendingRequests = failedRequests
this.saveToStorage()
httpRequest.destroy()
}
private saveToStorage(): void {
// 将待同步请求持久化
}
}8.3 7.3 Serverless与云函数
8.3.1 7.3.1 华为云Serverless架构
华为云FunctionGraph提供了Serverless计算服务，开发者无需管理服务器即可运行代码。在HarmonyOS应用中，可以通过华为云Serverless能力实现端云一体化开发，降低后端运维成本。
Serverless架构的核心优势： - 零运维：无需管理服务器、操作系统和中间件 - 弹性伸缩：自动根据请求量扩缩容 - 按需付费：仅按实际执行的计算资源计费 - 事件驱动：支持多种触发器（HTTP、定时、消息队列等）
8.3.2 7.3.2 云函数开发与部署
云函数是Serverless架构的核心单元。以下是一个典型的云函数示例：
// 云函数示例：用户数据聚合处理
// 部署在华为云FunctionGraph上
export async function handler(event: object, context: object): Promise<object> {
const requestBody = JSON.parse(JSON.stringify(event))
const { userId, dataType } = requestBody
try {
// 根据数据类型执行不同的聚合逻辑
switch (dataType) {
case 'profile':
return await aggregateUserProfile(userId)
case 'statistics':
return await aggregateStatistics(userId)
case 'notifications':
return await getNotifications(userId)
default:
return { code: 400, message: '不支持的数据类型' }
}
} catch (error) {
return { code: 500, message: '服务器内部错误' }
}
}
async function aggregateUserProfile(userId: string): Promise<object> {
// 从云数据库获取用户数据
// 执行聚合计算
// 返回结果
return {
code: 200,
data: {
userId: userId,
name: '张三',
level: 5,
score: 9800
}
}
}
async function aggregateStatistics(userId: string): Promise<object> {
return {
code: 200,
data: {
totalOrders: 156,
totalSpent: 28900,
averageRating: 4.8
}
}
}
async function getNotifications(userId: string): Promise<object> {
return {
code: 200,
data: {
unread: 3,
items: [
{ id: 1, title: '系统更新', time: '2024-01-15' },
{ id: 2, title: '活动通知', time: '2024-01-14' },
{ id: 3, title: '好友消息', time: '2024-01-13' }
]
}
}
}8.3.3 7.3.3 端侧调用云函数
HarmonyOS提供了AGConnect SDK，支持端侧直接调用华为云云函数。
import { functionGraph } from '@kit.CloudFoundationKit'
// 云函数调用封装
class CloudFunctionClient {
private static instance: CloudFunctionClient | null = null
static getInstance(): CloudFunctionClient {
if (!CloudFunctionClient.instance) {
CloudFunctionClient.instance = new CloudFunctionClient()
}
return CloudFunctionClient.instance
}
// 调用云函数
async invoke<T>(functionName: string, params: object): Promise<T> {
try {
const result = await functionGraph.callFunction(functionName, params)
return result.returnValue as T
} catch (error) {
console.error(`云函数 ${functionName} 调用失败: ${(error as Error).message}`)
throw error
}
}
// 调用云函数（带超时控制）
async invokeWithTimeout<T>(
functionName: string,
params: object,
timeoutMs: number = 10000
): Promise<T> {
return Promise.race([
this.invoke<T>(functionName, params),
new Promise<T>((_, reject) =>
setTimeout(() => reject(new Error('云函数调用超时')), timeoutMs)
)
])
}
}
// 使用示例
@Component
struct UserDashboard {
@State userProfile: UserProfile | null = null
@State isLoading: boolean = true
aboutToAppear(): void {
this.loadUserData()
}
async loadUserData(): Promise<void> {
try {
this.isLoading = true
const client = CloudFunctionClient.getInstance()
this.userProfile = await client.invoke<UserProfile>('getUserData', {
userId: 'current_user',
dataType: 'profile'
})
} catch (error) {
// 降级到本地缓存
this.userProfile = await this.loadFromCache()
} finally {
this.isLoading = false
}
}
async loadFromCache(): Promise<UserProfile | null> {
// 从本地缓存加载
return null
}
build() {
Column() {
if (this.isLoading) {
LoadingProgress()
} else if (this.userProfile) {
Text(this.userProfile.name)
.fontSize(24)
.fontWeight(FontWeight.Bold)
}
}
}
}8.3.4 7.3.4 云数据库集成
华为云数据库服务为HarmonyOS应用提供了云端数据持久化能力。
// 云数据库操作封装
class CloudDatabase {
private tableName: string
constructor(tableName: string) {
this.tableName = tableName
}
// 插入数据
async insert(record: object): Promise<string> {
const client = CloudFunctionClient.getInstance()
const result = await client.invoke<{ id: string }>('databaseOperation', {
action: 'insert',
table: this.tableName,
data: {
...record,
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString()
}
})
return result.id
}
// 查询数据
async query(filter: Record<string, object>, limit: number = 20): Promise<object[]> {
const client = CloudFunctionClient.getInstance()
const result = await client.invoke<{ items: object[] }>('databaseOperation', {
action: 'query',
table: this.tableName,
filter: filter,
limit: limit
})
return result.items
}
// 更新数据
async update(id: string, data: object): Promise<void> {
const client = CloudFunctionClient.getInstance()
await client.invoke('databaseOperation', {
action: 'update',
table: this.tableName,
id: id,
data: {
...data,
updatedAt: new Date().toISOString()
}
})
}
// 删除数据
async delete(id: string): Promise<void> {
const client = CloudFunctionClient.getInstance()
await client.invoke('databaseOperation', {
action: 'delete',
table: this.tableName,
id: id
})
}
}8.3.5 7.3.5 云存储集成
云存储服务用于管理应用中的文件资源，如用户上传的图片、视频、文档等。
// 云存储操作封装
class CloudStorage {
// 上传文件
async uploadFile(filePath: string, cloudPath: string): Promise<string> {
const client = CloudFunctionClient.getInstance()
// 获取上传凭证
const uploadToken = await client.invoke<{
uploadUrl: string,
token: string
}>('getUploadToken', {
fileName: cloudPath,
contentType: this.getContentType(filePath)
})
// 使用凭证上传文件
const httpRequest = http.createHttp()
try {
const response = await httpRequest.request(uploadToken.uploadUrl, {
method: http.RequestMethod.PUT,
header: {
'Authorization': uploadToken.token,
'Content-Type': this.getContentType(filePath)
},
extraData: filePath // 实际开发中传入文件二进制数据
})
if (response.responseCode === http.ResponseCode.OK ||
response.responseCode === http.ResponseCode.NO_CONTENT) {
return cloudPath
}
throw new Error('文件上传失败')
} finally {
httpRequest.destroy()
}
}
// 获取文件下载URL
async getDownloadUrl(cloudPath: string, expireSeconds: number = 3600): Promise<string> {
const client = CloudFunctionClient.getInstance()
const result = await client.invoke<{ url: string }>('getDownloadUrl', {
filePath: cloudPath,
expireSeconds: expireSeconds
})
return result.url
}
// 删除文件
async deleteFile(cloudPath: string): Promise<void> {
const client = CloudFunctionClient.getInstance()
await client.invoke('deleteFile', { filePath: cloudPath })
}
private getContentType(filePath: string): string {
const ext = filePath.split('.').pop()?.toLowerCase()
const mimeTypes: Record<string, string> = {
'jpg': 'image/jpeg', 'jpeg': 'image/jpeg',
'png': 'image/png', 'gif': 'image/gif',
'mp4': 'video/mp4', 'pdf': 'application/pdf',
'json': 'application/json', 'txt': 'text/plain'
}
return mimeTypes[ext ?? ''] ?? 'application/octet-stream'
}
}8.4 7.4 数据同步策略
8.4.1 7.4.1 增量同步与全量同步
数据同步是端云协同的核心挑战之一。根据数据量和变化频率，可选择不同的同步策略。
全量同步适用于数据量小、变化频繁的场景，每次同步传输完整数据集。
增量同步适用于数据量大、变化较少的场景，仅传输自上次同步以来发生变化的数据。
// 同步管理器
class SyncManager {
private lastSyncTime: number = 0
private syncLock: boolean = false
// 全量同步
async fullSync(tableName: string): Promise<SyncResult> {
if (this.syncLock) {
throw new Error('同步正在进行中')
}
this.syncLock = true
try {
const client = CloudFunctionClient.getInstance()
const cloudData = await client.invoke<{
items: object[],
serverTime: number
}>('fullSync', { table: tableName })
// 用云端数据覆盖本地数据
await LocalDatabase.getInstance().replaceTable(tableName, cloudData.items)
this.lastSyncTime = cloudData.serverTime
return {
success: true,
syncedCount: cloudData.items.length,
syncTime: cloudData.serverTime
}
} catch (error) {
return {
success: false,
syncedCount: 0,
syncTime: 0,
error: (error as Error).message
}
} finally {
this.syncLock = false
}
}
// 增量同步
async incrementalSync(tableName: string): Promise<SyncResult> {
if (this.syncLock) {
throw new Error('同步正在进行中')
}
this.syncLock = true
try {
const client = CloudFunctionClient.getInstance()
// 上传本地变更
const localChanges = await LocalDatabase.getInstance()
.getChangesSince(tableName, this.lastSyncTime)
// 获取云端变更
const result = await client.invoke<{
updates: object[],
deletes: string[],
serverTime: number
}>('incrementalSync', {
table: tableName,
since: this.lastSyncTime,
localChanges: localChanges
})
// 应用云端变更到本地
await this.applyCloudChanges(tableName, result)
this.lastSyncTime = result.serverTime
return {
success: true,
syncedCount: result.updates.length + result.deletes.length,
syncTime: result.serverTime
}
} catch (error) {
return {
success: false,
syncedCount: 0,
syncTime: 0,
error: (error as Error).message
}
} finally {
this.syncLock = false
}
}
private async applyCloudChanges(
tableName: string,
changes: { updates: object[], deletes: string[] }
): Promise<void> {
const db = LocalDatabase.getInstance()
// 应用更新
for (const item of changes.updates) {
await db.upsert(tableName, item)
}
// 应用删除
for (const id of changes.deletes) {
await db.delete(tableName, id)
}
// 清除已同步的本地变更记录
await db.clearChanges(tableName)
}
}
interface SyncResult {
success: boolean
syncedCount: number
syncTime: number
error?: string
}8.4.2 7.4.2 冲突解决策略
当端侧和云侧同时修改了同一条数据时，会产生冲突。常见的冲突解决策略包括：
- 云端优先（Server Wins）：以云端数据为准，丢弃本地修改
- 本地优先（Client Wins）：以本地数据为准，覆盖云端
- 时间戳优先（Last Write Wins）：以最后修改时间为准
- 字段级合并（Field-level Merge）：对不同字段的修改分别保留
- 自定义合并（Custom Merge）：根据业务逻辑自定义合并规则
// 冲突解决策略接口
interface ConflictResolver {
resolve(localRecord: object, cloudRecord: object): object
}
// 时间戳优先策略
class LastWriteWinsResolver implements ConflictResolver {
resolve(localRecord: object, cloudRecord: object): object {
const localTime = (localRecord as Record<string, string>)['updatedAt']
const cloudTime = (cloudRecord as Record<string, string>)['updatedAt']
if (localTime > cloudTime) {
return localRecord
}
return cloudRecord
}
}
// 字段级合并策略
class FieldLevelMergeResolver implements ConflictResolver {
private priorityFields: string[]
constructor(priorityFields: string[] = ['cloud']) {
this.priorityFields = priorityFields
}
resolve(localRecord: object, cloudRecord: object): object {
const merged = { ...cloudRecord }
const local = localRecord as Record<string, object>
const cloud = cloudRecord as Record<string, object>
for (const key of Object.keys(local)) {
if (key === 'updatedAt' || key === 'createdAt') continue
if (this.priorityFields.includes(key)) {
// 优先字段以云端为准
merged[key] = cloud[key]
} else if (local[key] !== cloud[key]) {
// 非优先字段取最后修改的
const localModified = (localRecord as Record<string, string>)[`${key}_modifiedAt`]
const cloudModified = (cloudRecord as Record<string, string>)[`${key}_modifiedAt`]
if (localModified && cloudModified) {
merged[key] = localModified > cloudModified ? local[key] : cloud[key]
} else {
merged[key] = local[key] // 默认取本地
}
}
}
return merged
}
}
// 带冲突检测的同步
class ConflictAwareSync {
private resolver: ConflictResolver
constructor(resolver: ConflictResolver) {
this.resolver = resolver
}
async syncWithConflictResolution(
tableName: string,
localChanges: object[],
cloudChanges: object[]
): Promise<object[]> {
const resolvedChanges: object[] = []
const localMap = new Map<string, object>()
// 构建本地变更索引
for (const change of localChanges) {
const id = (change as Record<string, string>)['id']
localMap.set(id, change)
}
// 检测并解决冲突
for (const cloudChange of cloudChanges) {
const id = (cloudChange as Record<string, string>)['id']
const localChange = localMap.get(id)
if (localChange) {
// 存在冲突，执行解决策略
const resolved = this.resolver.resolve(localChange, cloudChange)
resolvedChanges.push(resolved)
localMap.delete(id)
} else {
resolvedChanges.push(cloudChange)
}
}
// 未冲突的本地变更直接保留
for (const localChange of localMap.values()) {
resolvedChanges.push(localChange)
}
return resolvedChanges
}
}8.4.3 7.4.3 离线优先架构
离线优先（Offline-First）是一种将离线作为默认状态的设计模式。应用在没有网络的情况下也能正常工作，网络恢复后自动同步数据。
// 离线优先数据管理器
class OfflineFirstDataManager {
private localDb: LocalDatabase
private syncManager: SyncManager
private networkMonitor: NetworkMonitor
constructor() {
this.localDb = LocalDatabase.getInstance()
this.syncManager = new SyncManager()
this.networkMonitor = NetworkMonitor.getInstance()
// 监听网络恢复事件
this.networkMonitor.onNetworkChange(async (online: boolean) => {
if (online) {
console.info('网络恢复，开始同步数据')
await this.syncAllTables()
}
})
}
// 读取数据（优先本地）
async read<T>(tableName: string, id: string): Promise<T | null> {
// 始终从本地读取，保证响应速度
const localData = await this.localDb.getById<T>(tableName, id)
// 如果在线，后台静默更新
if (this.networkMonitor.getOnlineStatus()) {
this.silentRefresh(tableName, id)
}
return localData
}
// 写入数据（本地优先）
async write(tableName: string, data: object): Promise<void> {
// 先写入本地
const record = {
...data,
_localModified: Date.now(),
_syncStatus: 'pending'
}
await this.localDb.upsert(tableName, record)
// 如果在线，尝试立即同步
if (this.networkMonitor.getOnlineStatus()) {
try {
await this.syncSingleRecord(tableName, record)
} catch (error) {
console.warn('即时同步失败，等待下次同步')
}
}
}
// 静默刷新单条数据
private async silentRefresh(tableName: string, id: string): Promise<void> {
try {
const client = CloudFunctionClient.getInstance()
const cloudData = await client.invoke<object>('getRecord', {
table: tableName,
id: id
})
if (cloudData) {
await this.localDb.upsert(tableName, cloudData)
}
} catch (error) {
// 静默失败，不影响用户体验
}
}
// 同步所有表
private async syncAllTables(): Promise<void> {
const tables = await this.localDb.getAllTables()
for (const table of tables) {
try {
await this.syncManager.incrementalSync(table)
} catch (error) {
console.error(`表 ${table} 同步失败`)
}
}
}
// 同步单条记录
private async syncSingleRecord(tableName: string, record: object): Promise<void> {
const client = CloudFunctionClient.getInstance()
await client.invoke('upsertRecord', {
table: tableName,
data: record
})
// 更新同步状态
await this.localDb.updateSyncStatus(tableName, record['id'], 'synced')
}
}8.4.4 7.4.4 数据缓存策略
合理的缓存策略可以显著减少网络请求次数，提升应用响应速度。
// 多级缓存管理器
class CacheManager {
private memoryCache: Map<string, CacheEntry> = new Map()
private maxMemorySize: number = 100 // 最大缓存条目数
private defaultTTL: number = 5 * 60 * 1000 // 默认5分钟过期
// 获取缓存
async get<T>(key: string): Promise<T | null> {
// 1. 先查内存缓存
const memEntry = this.memoryCache.get(key)
if (memEntry && !this.isExpired(memEntry)) {
memEntry.accessCount++
memEntry.lastAccessTime = Date.now()
return memEntry.data as T
}
// 2. 查持久化缓存
const persistedData = await this.getPersistedCache<T>(key)
if (persistedData) {
// 回填内存缓存
this.memoryCache.set(key, {
data: persistedData,
createTime: Date.now(),
expireTime: Date.now() + this.defaultTTL,
accessCount: 1,
lastAccessTime: Date.now()
})
return persistedData
}
return null
}
// 设置缓存
async set<T>(key: string, value: T, ttl?: number): Promise<void> {
const entry: CacheEntry = {
data: value,
createTime: Date.now(),
expireTime: Date.now() + (ttl ?? this.defaultTTL),
accessCount: 0,
lastAccessTime: Date.now()
}
// 内存缓存满时执行淘汰
if (this.memoryCache.size >= this.maxMemorySize) {
this.evictLRU()
}
this.memoryCache.set(key, entry)
// 同时写入持久化缓存
await this.setPersistedCache(key, value, ttl)
}
// 缓存失效
invalidate(key: string): void {
this.memoryCache.delete(key)
// 同时清除持久化缓存
this.removePersistedCache(key)
}
// LRU淘汰
private evictLRU(): void {
let oldestKey: string | null = null
let oldestTime = Infinity
for (const [key, entry] of this.memoryCache) {
if (entry.lastAccessTime < oldestTime) {
oldestTime = entry.lastAccessTime
oldestKey = key
}
}
if (oldestKey) {
this.memoryCache.delete(oldestKey)
}
}
private isExpired(entry: CacheEntry): boolean {
return Date.now() > entry.expireTime
}
private async getPersistedCache<T>(key: string): Promise<T | null> {
// 从Preferences或RDB读取
return null
}
private async setPersistedCache<T>(key: string, value: T, ttl?: number): Promise<void> {
// 写入Preferences或RDB
}
private removePersistedCache(key: string): void {
// 删除持久化缓存
}
}
interface CacheEntry {
data: object
createTime: number
expireTime: number
accessCount: number
lastAccessTime: number
}8.4.5 7.4.5 同步状态管理
同步状态管理用于跟踪数据在端云之间的同步状态，确保数据一致性。
// 同步状态枚举
enum SyncStatus {
SYNCED = 'synced', // 已同步
PENDING = 'pending', // 待同步
SYNCING = 'syncing', // 同步中
CONFLICT = 'conflict', // 冲突
FAILED = 'failed' // 同步失败
}
// 同步状态管理器
class SyncStateManager {
private statusMap: Map<string, SyncStatus> = new Map()
private statusListeners: Array<(id: string, status: SyncStatus) => void> = []
updateStatus(recordId: string, status: SyncStatus): void {
this.statusMap.set(recordId, status)
this.notifyListeners(recordId, status)
}
getStatus(recordId: string): SyncStatus {
return this.statusMap.get(recordId) ?? SyncStatus.SYNCED
}
onStatusChange(listener: (id: string, status: SyncStatus) => void): void {
this.statusListeners.push(listener)
}
getPendingCount(): number {
let count = 0
for (const status of this.statusMap.values()) {
if (status === SyncStatus.PENDING || status === SyncStatus.FAILED) {
count++
}
}
return count
}
private notifyListeners(id: string, status: SyncStatus): void {
this.statusListeners.forEach(listener => listener(id, status))
}
}8.5 7.5 推送与实时通信
8.5.1 7.5.1 华为推送服务集成
华为推送服务（Push Kit）提供了从云端到终端的消息推送能力，支持通知栏消息、透传消息等多种形式。
import { pushService } from '@kit.PushKit'
// 推送服务管理器
class PushManager {
private static instance: PushManager | null = null
static getInstance(): PushManager {
if (!PushManager.instance) {
PushManager.instance = new PushManager()
}
return PushManager.instance
}
// 初始化推送服务
async init(): Promise<void> {
try {
// 申请推送权限
const result = await pushService.requestEnableNotification()
if (result === 0) {
console.info('推送通知已启用')
}
// 获取推送Token
const tokenResult = await pushService.getToken()
if (tokenResult.resultCode === 0) {
console.info(`推送Token: ${tokenResult.token}`)
// 将Token上报到业务服务器
await this.reportTokenToServer(tokenResult.token)
}
} catch (error) {
console.error(`推送服务初始化失败: ${(error as Error).message}`)
}
}
// 注册推送消息回调
registerMessageHandler(): void {
// 处理透传消息
pushService.on('pushMessage', (data: pushService.PushPayload) => {
console.info('收到透传消息')
this.handlePushMessage(data)
})
}
// 处理推送消息
private handlePushMessage(data: pushService.PushPayload): void {
const message = data.data
if (message) {
try {
const parsed = JSON.parse(message)
// 根据消息类型执行不同处理
switch (parsed.type) {
case 'chat':
this.handleChatMessage(parsed)
break
case 'update':
this.handleUpdateNotification(parsed)
break
case 'promotion':
this.handlePromotion(parsed)
break
}
} catch (e) {
console.error('消息解析失败')
}
}
}
private handleChatMessage(message: object): void {
// 处理聊天消息
}
private handleUpdateNotification(message: object): void {
// 处理更新通知
}
private handlePromotion(message: object): void {
// 处理营销消息
}
private async reportTokenToServer(token: string): Promise<void> {
// 上报Token到业务服务器
}
}8.5.2 7.5.2 WebSocket长连接
WebSocket提供了全双工的实时通信能力，适用于聊天、实时协作等场景。
// 实时通信管理器
class RealtimeManager {
private wsClient: WebSocketClient
private messageQueue: Array<{ type: string; data: object }> = []
private isConnected: boolean = false
constructor() {
this.wsClient = new WebSocketClient()
}
async connect(serverUrl: string): Promise<void> {
await this.wsClient.connect(serverUrl)
this.isConnected = true
// 注册消息处理器
this.wsClient.on('chat_message', (data: object) => {
this.onChatMessage(data)
})
this.wsClient.on('typing_indicator', (data: object) => {
this.onTypingIndicator(data)
})
this.wsClient.on('presence', (data: object) => {
this.onPresenceChange(data)
})
// 发送队列中的待发消息
this.flushMessageQueue()
}
// 发送聊天消息
sendChatMessage(toUserId: string, content: string, messageType: string = 'text'): void {
const message = {
toUserId: toUserId,
content: content,
messageType: messageType,
timestamp: Date.now(),
clientId: this.generateClientId()
}
if (this.isConnected) {
this.wsClient.send('chat_message', message)
} else {
// 离线时加入队列
this.messageQueue.push({ type: 'chat_message', data: message })
}
}
// 发送输入状态
sendTypingIndicator(conversationId: string): void {
if (this.isConnected) {
this.wsClient.send('typing_indicator', {
conversationId: conversationId,
timestamp: Date.now()
})
}
}
private flushMessageQueue(): void {
while (this.messageQueue.length > 0) {
const msg = this.messageQueue.shift()!
this.wsClient.send(msg.type, msg.data)
}
}
private generateClientId(): string {
return `msg_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`
}
private onChatMessage(data: object): void {
// 处理收到的聊天消息
}
private onTypingIndicator(data: object): void {
// 处理输入状态指示
}
private onPresenceChange(data: object): void {
// 处理在线状态变化
}
disconnect(): void {
this.wsClient.disconnect()
this.isConnected = false
}
}8.5.3 7.5.3 消息推送策略
合理的消息推送策略可以平衡实时性和设备功耗。
// 推送策略管理器
class PushStrategyManager {
// 根据消息优先级选择推送通道
getPushChannel(messagePriority: string): 'push' | 'websocket' | 'polling' {
switch (messagePriority) {
case 'critical': // 紧急消息（如安全告警）
return 'push' // 使用系统推送，确保即时送达
case 'high': // 高优先级（如聊天消息）
return 'websocket' // 使用WebSocket长连接
case 'normal': // 普通消息（如动态更新）
return 'websocket'
case 'low': // 低优先级（如营销消息）
return 'polling' // 使用轮询，节省资源
default:
return 'polling'
}
}
// 智能轮询间隔
getPollingInterval(userActivity: string): number {
switch (userActivity) {
case 'active': return 10000 // 活跃状态：10秒
case 'foreground': return 30000 // 前台：30秒
case 'background': return 300000 // 后台：5分钟
case 'idle': return 900000 // 空闲：15分钟
default: return 60000 // 默认：1分钟
}
}
}8.5.4 7.5.4 实时消息处理
// 实时消息处理器
class MessageProcessor {
private messageHandlers: Map<string, MessageHandler> = new Map()
private messageStore: MessageStore
constructor() {
this.messageStore = new MessageStore()
}
registerHandler(type: string, handler: MessageHandler): void {
this.messageHandlers.set(type, handler)
}
async processMessage(rawMessage: string): Promise<void> {
try {
const message: AppMessage = JSON.parse(rawMessage)
// 消息去重
if (await this.messageStore.isDuplicate(message.id)) {
console.info(`消息 ${message.id} 已处理，跳过`)
return
}
// 消息存储
await this.messageStore.save(message)
// 分发到对应处理器
const handler = this.messageHandlers.get(message.type)
if (handler) {
await handler.handle(message)
} else {
console.warn(`未找到消息类型 ${message.type} 的处理器`)
}
// 发送确认
this.sendAck(message.id)
} catch (error) {
console.error(`消息处理失败: ${(error as Error).message}`)
}
}
private sendAck(messageId: string): void {
// 发送消息确认
}
}
interface MessageHandler {
handle(message: AppMessage): Promise<void>
}
interface AppMessage {
id: string
type: string
data: object
timestamp: number
from: string
}
class MessageStore {
async isDuplicate(messageId: string): Promise<boolean> {
// 检查消息是否已处理
return false
}
async save(message: AppMessage): Promise<void> {
// 持久化消息
}
}8.6 7.6 端云一体化开发实践
8.6.1 7.6.1 完整项目架构设计
一个典型的端云一体化HarmonyOS应用包含以下层次：
项目结构：
├── entry/ # 应用入口模块
│ └── src/main/ets/
│ ├── entryability/ # Ability层
│ ├── pages/ # 页面层
│ ├── components/ # 组件层
│ ├── viewmodels/ # ViewModel层
│ ├── services/ # 服务层
│ │ ├── network/ # 网络服务
│ │ ├── storage/ # 存储服务
│ │ └── sync/ # 同步服务
│ ├── models/ # 数据模型
│ └── utils/ # 工具类
├── common/ # 公共模块
│ └── src/main/ets/
│ ├── constants/ # 常量定义
│ ├── types/ # 类型定义
│ └── base/ # 基类
└── cloudfunctions/ # 云函数
├── user-service/
├── data-service/
└── notification-service/8.6.2 7.6.2 API网关设计
API网关是端云通信的统一入口，负责路由、鉴权、限流、日志等职责。
// API网关客户端
class ApiGateway {
private baseUrl: string
private httpClient: HttpClient
private cacheManager: CacheManager
constructor() {
this.baseUrl = 'https://gateway.example.com'
this.httpClient = new HttpClient()
this.cacheManager = new CacheManager()
// 配置拦截器
this.httpClient.addRequestInterceptor(new AuthInterceptor())
this.httpClient.addRequestInterceptor(new SignatureInterceptor())
this.httpClient.addResponseInterceptor(new ErrorResponseInterceptor())
}
// 通用请求方法（带缓存）
async request<T>(
method: string,
path: string,
options?: {
body?: object,
cache?: boolean,
cacheTTL?: number,
retryCount?: number
}
): Promise<T> {
const cacheKey = `${method}:${path}:${JSON.stringify(options?.body ?? {})}`
// GET请求且启用缓存时，先查缓存
if (method === 'GET' && options?.cache) {
const cached = await this.cacheManager.get<T>(cacheKey)
if (cached) {
return cached
}
}
// 带重试的请求
const retryable = new RetryableRequest(options?.retryCount ?? 2)
const result = await retryable.execute(() =>
this.httpClient.request<T>({
url: `${this.baseUrl}${path}`,
method: method as http.RequestMethod,
headers: { 'Content-Type': 'application/json' },
body: options?.body ? JSON.stringify(options.body) : undefined
})
)
// 缓存GET请求结果
if (method === 'GET' && options?.cache) {
await this.cacheManager.set(cacheKey, result, options.cacheTTL)
}
return result
}
}
// 请求签名拦截器
class SignatureInterceptor implements RequestInterceptor {
private appId: string = 'your_app_id'
private appSecret: string = 'your_app_secret'
async intercept(config: RequestConfig): Promise<RequestConfig> {
const timestamp = Date.now().toString()
const nonce = Math.random().toString(36).substr(2, 16)
const signString = `${config.method}${config.url}${timestamp}${nonce}${this.appSecret}`
const signature = this.sha256(signString)
config.headers['X-App-Id'] = this.appId
config.headers['X-Timestamp'] = timestamp
config.headers['X-Nonce'] = nonce
config.headers['X-Signature'] = signature
return config
}
private sha256(data: string): string {
// 使用crypto API计算SHA256
return 'computed_signature'
}
}
// 错误响应拦截器
class ErrorResponseInterceptor implements ResponseInterceptor {
intercept(response: ResponseResult): ResponseResult {
if (response.code !== 200) {
switch (response.code) {
case 401:
// Token过期，触发重新登录
EventBus.getInstance().emit('auth_expired')
break
case 403:
// 权限不足
console.error('权限不足')
break
case 429:
// 请求过于频繁
console.warn('请求限流')
break
case 500:
// 服务器错误
console.error('服务器内部错误')
break
}
}
return response
}
interceptError(error: Error): void {
console.error(`网络请求异常: ${error.message}`)
}
}8.6.3 7.6.3 用户认证与授权
用户认证是端云协同中的安全基石。以下是基于JWT的认证方案：
// Token管理器
class TokenManager {
private static instance: TokenManager | null = null
private accessToken: string = ''
private refreshToken: string = ''
private tokenExpireTime: number = 0
static getInstance(): TokenManager {
if (!TokenManager.instance) {
TokenManager.instance = new TokenManager()
}
return TokenManager.instance
}
// 登录获取Token
async login(username: string, password: string): Promise<boolean> {
try {
const client = new HttpClient()
const result = await client.request<{
accessToken: string,
refreshToken: string,
expiresIn: number
}>({
url: 'https://auth.example.com/login',
method: http.RequestMethod.POST,
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ username, password })
})
this.accessToken = result.accessToken
this.refreshToken = result.refreshToken
this.tokenExpireTime = Date.now() + result.expiresIn * 1000
// 持久化Token
await this.persistTokens()
return true
} catch (error) {
console.error(`登录失败: ${(error as Error).message}`)
return false
}
}
// 获取有效Token（自动刷新）
async getValidToken(): Promise<string> {
// Token即将过期（提前5分钟刷新）
if (Date.now() > this.tokenExpireTime - 5 * 60 * 1000) {
await this.refreshAccessToken()
}
return this.accessToken
}
// 刷新Token
private async refreshAccessToken(): Promise<void> {
try {
const client = new HttpClient()
const result = await client.request<{
accessToken: string,
expiresIn: number
}>({
url: 'https://auth.example.com/refresh',
method: http.RequestMethod.POST,
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ refreshToken: this.refreshToken })
})
this.accessToken = result.accessToken
this.tokenExpireTime = Date.now() + result.expiresIn * 1000
await this.persistTokens()
} catch (error) {
// 刷新失败，需要重新登录
EventBus.getInstance().emit('auth_expired')
}
}
private async persistTokens(): Promise<void> {
// 使用安全存储保存Token
}
async logout(): Promise<void> {
this.accessToken = ''
this.refreshToken = ''
this.tokenExpireTime = 0
// 清除持久化Token
}
}8.6.4 7.6.4 日志与监控
完善的日志和监控体系是保障应用稳定运行的基础。
// 统一日志管理器
class LogManager {
private static instance: LogManager | null = null
private logBuffer: LogEntry[] = []
private maxBufferSize: number = 100
private flushInterval: number = 60000 // 60秒上报一次
static getInstance(): LogManager {
if (!LogManager.instance) {
LogManager.instance = new LogManager()
LogManager.instance.startAutoFlush()
}
return LogManager.instance
}
info(tag: string, message: string, extra?: object): void {
this.log(LogLevel.INFO, tag, message, extra)
}
warn(tag: string, message: string, extra?: object): void {
this.log(LogLevel.WARN, tag, message, extra)
}
error(tag: string, message: string, error?: Error, extra?: object): void {
this.log(LogLevel.ERROR, tag, message, { ...extra, error: error?.message })
}
private log(level: LogLevel, tag: string, message: string, extra?: object): void {
const entry: LogEntry = {
level: level,
tag: tag,
message: message,
timestamp: Date.now(),
extra: extra,
deviceId: 'device_id',
appVersion: '1.0.0'
}
this.logBuffer.push(entry)
// 本地日志输出
switch (level) {
case LogLevel.INFO: console.info(`[${tag}] ${message}`); break
case LogLevel.WARN: console.warn(`[${tag}] ${message}`); break
case LogLevel.ERROR: console.error(`[${tag}] ${message}`); break
}
// 缓冲区满时自动上报
if (this.logBuffer.length >= this.maxBufferSize) {
this.flush()
}
}
async flush(): Promise<void> {
if (this.logBuffer.length === 0) return
const logsToUpload = [...this.logBuffer]
this.logBuffer = []
try {
// 批量上报日志到云端
const client = CloudFunctionClient.getInstance()
await client.invoke('uploadLogs', { logs: logsToUpload })
} catch (error) {
// 上报失败，将日志放回缓冲区
this.logBuffer = [...logsToUpload, ...this.logBuffer]
}
}
private startAutoFlush(): void {
setInterval(() => this.flush(), this.flushInterval)
}
}
enum LogLevel {
INFO = 'INFO',
WARN = 'WARN',
ERROR = 'ERROR'
}
interface LogEntry {
level: LogLevel
tag: string
message: string
timestamp: number
extra?: object
deviceId: string
appVersion: string
}8.6.5 7.6.5 灰度发布与版本管理
灰度发布允许将新版本逐步推送给部分用户，降低发布风险。
// 版本管理器
class VersionManager {
private static instance: VersionManager | null = null
static getInstance(): VersionManager {
if (!VersionManager.instance) {
VersionManager.instance = new VersionManager()
}
return VersionManager.instance
}
// 检查更新
async checkUpdate(): Promise<UpdateInfo | null> {
try {
const client = CloudFunctionClient.getInstance()
const result = await client.invoke<UpdateInfo>('checkUpdate', {
currentVersion: this.getCurrentVersion(),
deviceId: this.getDeviceId(),
userId: this.getUserId(),
channel: this.getChannel()
})
if (result.hasUpdate) {
return result
}
return null
} catch (error) {
console.error('检查更新失败')
return null
}
}
// 灰度判断
async isInGrayGroup(): Promise<boolean> {
const client = CloudFunctionClient.getInstance()
const result = await client.invoke<{ inGray: boolean }>('checkGray', {
deviceId: this.getDeviceId(),
currentVersion: this.getCurrentVersion()
})
return result.inGray
}
// 动态配置下发
async getRemoteConfig(): Promise<Record<string, string>> {
const client = CloudFunctionClient.getInstance()
const result = await client.invoke<{
config: Record<string, string>
}>('getRemoteConfig', {
appVersion: this.getCurrentVersion(),
userId: this.getUserId()
})
return result.config
}
private getCurrentVersion(): string {
return '1.0.0'
}
private getDeviceId(): string {
return 'mock_device_id'
}
private getUserId(): string {
return 'mock_user_id'
}
private getChannel(): string {
return 'official'
}
}
interface UpdateInfo {
hasUpdate: boolean
version: string
downloadUrl: string
changelog: string
forceUpdate: boolean
grayPercent: number
}8.7 本章小结
本章深入讲解了HarmonyOS端云协同开发的完整知识体系：
端云协同架构：理解了B/S、C/S和混合架构的特点与适用场景，掌握了端云职责划分的基本原则和通信协议的选择策略。
网络请求管理：深入学习了HTTP模块的使用，掌握了请求/响应拦截器的实现、请求队列与并发控制、超时重试策略以及网络状态监听。
Serverless与云函数：了解了华为云Serverless架构，掌握了云函数的开发部署、端侧调用云函数以及云数据库和云存储的集成方法。
数据同步策略：深入理解了增量同步与全量同步的区别，掌握了冲突解决策略、离线优先架构、多级缓存策略和同步状态管理。
推送与实时通信：掌握了华为推送服务的集成、WebSocket长连接的实现、消息推送策略的设计以及实时消息的处理流程。
端云一体化实践：通过完整的项目架构设计、API网关、用户认证、日志监控和灰度发布等实践，形成了端云协同开发的完整方法论。
8.8 练习题
8.8.1 一、单选题
1. 在HarmonyOS端云协同架构中，以下哪种通信协议最适合实时聊天场景？
A. HTTP/1.1 B. WebSocket C. FTP D. SMTP
答案：B
解析： WebSocket提供全双工通信能力，支持服务端主动推送消息，非常适合实时聊天场景。HTTP/1.1是请求-响应模式，不适合实时双向通信；FTP用于文件传输；SMTP用于邮件传输。
2. 关于端云协同中的离线优先架构，以下说法正确的是？
A. 离线时所有功能都不可用 B. 数据优先写入本地，网络恢复后自动同步到云端 C. 离线时只能读取数据不能写入 D. 离线优先架构不需要考虑冲突解决
答案：B
解析： 离线优先架构的核心理念是数据优先写入本地存储，保证应用在无网络时仍可正常使用。当网络恢复后，自动将本地变更同步到云端。由于可能存在端云同时修改同一数据的情况，冲突解决是必须考虑的问题。
3. 在请求重试策略中，指数退避（Exponential Backoff）的主要目的是什么？
A. 加快请求速度 B. 减少服务器压力，避免雪崩效应 C. 增加请求成功率 D. 减少代码复杂度
答案：B
解析： 指数退避策略在每次重试时逐渐增加等待时间（如1s、2s、4s、8s…），目的是避免大量失败请求同时重试导致服务器压力过大，防止雪崩效应。
8.8.2 二、多选题
4. 以下哪些属于端云协同中常见的冲突解决策略？
A. 云端优先（Server Wins） B. 本地优先（Client Wins） C. 时间戳优先（Last Write Wins） D. 随机选择（Random Pick） E. 字段级合并（Field-level Merge）
答案：A、B、C、E
解析： 常见的冲突解决策略包括云端优先、本地优先、时间戳优先和字段级合并。随机选择不是一种合理的冲突解决策略，可能导致数据不一致或用户体验问题。
5. 关于HarmonyOS网络请求管理，以下哪些说法是正确的？
A. 使用完http.createHttp()创建的对象后应调用destroy()释放资源 B. connectTimeout和readTimeout分别控制连接超时和读取超时 C. HTTP请求只能在主线程中发起 D. 可以通过header参数设置请求头 E. extraData参数用于设置POST请求体
答案：A、B、D、E
解析： HarmonyOS中HTTP请求对象使用完毕后必须调用destroy()释放资源；connectTimeout和readTimeout分别控制连接和读取超时；HTTP请求可以在子线程中发起，不会阻塞主线程；header参数设置请求头，extraData设置请求体。
8.8.3 三、判断题
6. Serverless架构意味着应用完全不需要服务器。
答案：错误
解析： Serverless并非不需要服务器，而是开发者无需管理服务器的运维工作。服务器由云服务提供商（如华为云）管理，开发者只需关注业务逻辑代码的编写和部署。
7. 在端云协同开发中，WebSocket连接断开后应立即以固定间隔无限重试重连。
答案：错误
解析： WebSocket重连应采用指数退避策略并设置最大重试次数，避免在服务端不可用时持续发起重连请求，造成资源浪费和服务端压力。无限重试可能导致设备电量和流量的无谓消耗。
8. 离线优先架构中，本地数据的写入速度通常比直接写入云端更快。
答案：正确
解析： 离线优先架构中，数据首先写入本地存储（如RDB、Preferences），本地存储的I/O速度远快于网络请求，因此写入速度更快。这也是离线优先架构能提供良好用户体验的重要原因之一。