8  第7章 端云协同开发

8.1 7.1 端云协同架构概述

8.1.1 7.1.1 端云协同的设计理念

端云协同开发是HarmonyOS应用开发中的核心能力之一。所谓”端云协同”,是指将终端设备(手机、平板、穿戴设备等)的计算能力与云端服务器的计算能力有机结合,通过高效的通信机制和协同策略,实现应用功能的最佳分配与数据的高效流转。

端云协同的核心设计理念包括:

  1. 以端侧体验为中心:云端能力服务于端侧用户体验,确保低延迟、高可靠的交互响应。
  2. 能力互补:端侧擅长实时交互、传感器数据采集和本地缓存;云侧擅长海量存储、复杂计算和多端数据汇聚。
  3. 弹性伸缩:应用逻辑可根据网络状况、设备性能等因素,在端侧和云侧之间灵活调度。
  4. 数据一致性:通过同步机制保证端云数据的最终一致性,支持离线场景下的数据完整性。

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 冲突解决策略

当端侧和云侧同时修改了同一条数据时,会产生冲突。常见的冲突解决策略包括:

  1. 云端优先(Server Wins):以云端数据为准,丢弃本地修改
  2. 本地优先(Client Wins):以本地数据为准,覆盖云端
  3. 时间戳优先(Last Write Wins):以最后修改时间为准
  4. 字段级合并(Field-level Merge):对不同字段的修改分别保留
  5. 自定义合并(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端云协同开发的完整知识体系:

  1. 端云协同架构:理解了B/S、C/S和混合架构的特点与适用场景,掌握了端云职责划分的基本原则和通信协议的选择策略。

  2. 网络请求管理:深入学习了HTTP模块的使用,掌握了请求/响应拦截器的实现、请求队列与并发控制、超时重试策略以及网络状态监听。

  3. Serverless与云函数:了解了华为云Serverless架构,掌握了云函数的开发部署、端侧调用云函数以及云数据库和云存储的集成方法。

  4. 数据同步策略:深入理解了增量同步与全量同步的区别,掌握了冲突解决策略、离线优先架构、多级缓存策略和同步状态管理。

  5. 推送与实时通信:掌握了华为推送服务的集成、WebSocket长连接的实现、消息推送策略的设计以及实时消息的处理流程。

  6. 端云一体化实践:通过完整的项目架构设计、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速度远快于网络请求,因此写入速度更快。这也是离线优先架构能提供良好用户体验的重要原因之一。