概述

在移动应用开发中，离线数据同步是一项至关重要的技术。用户在离线状态下进行的操作需要在网络恢复后自动同步到服务器，以确保数据的一致性和完整性。

Flutter提供了多种离线数据同步方案，可以在鸿蒙平台上实现可靠的数据同步机制。本文将深入探讨离线数据同步的核心概念、同步策略、实现方法以及在鸿蒙平台上的最佳实践。

核心概念

离线同步架构

┌─────────────────────────────────────────────────────────────┐
│                    离线同步架构                              │
├─────────────────────────────────────────────────────────────┤
│  用户操作层                                                   │
│  ┌───────────────────────────────────────┐                  │
│  │  增删改查操作                          │                  │
│  │  本地数据更新                          │                  │
│  └───────────────────────────────────────┘                  │
│              ↓                                                │
│  同步队列层                                                   │
│  ┌───────────────────────────────────────┐                  │
│  │  同步操作队列                          │                  │
│  │  优先级管理                            │                  │
│  │  重试机制                              │                  │
│  └───────────────────────────────────────┘                  │
│              ↓                                                │
│  网络状态层                                                   │
│  ┌───────────────────────────────────────┐                  │
│  │  网络状态监听                          │                  │
│  │  自动触发同步                          │                  │
│  │  网络质量评估                          │                  │
│  └───────────────────────────────────────┘                  │
│              ↓                                                │
│  数据传输层                                                   │
│  ┌───────────────────────────────────────┐                  │
│  │  HTTP请求                              │                  │
│  │  数据序列化                            │                  │
│  │  错误处理                              │                  │
│  └───────────────────────────────────────┘                  │
│              ↓                                                │
│  服务器层                                                     │
│  ┌───────────────────────────────────────┐                  │
│  │  数据接收与处理                        │                  │
│  │  冲突检测与解决                        │                  │
│  │  响应反馈                              │                  │
│  └───────────────────────────────────────┘                  │
└─────────────────────────────────────────────────────────────┘

同步状态

状态	描述	处理方式
pending	等待同步	网络恢复后自动同步
syncing	正在同步	等待完成
completed	同步完成	从队列中移除
failed	同步失败	重试或标记错误
skipped	跳过同步	记录并跳过

同步优先级

优先级	描述	示例
urgent	紧急	支付操作
high	高	用户登录/注册
normal	普通	数据更新
low	低	日志上报

基本使用

添加依赖

dependencies:
  flutter:
    sdk: flutter
  connectivity_plus: ^4.0.1
  sqflite: ^2.2.8
  json_annotation: ^4.8.1

同步配置

class SyncConfig {
  final int maxRetries;
  final Duration retryDelay;
  final int batchSize;
  final Duration syncInterval;
  final bool autoSyncOnNetworkChange;

  const SyncConfig({
    this.maxRetries = 3,
    this.retryDelay = const Duration(seconds: 5),
    this.batchSize = 10,
    this.syncInterval = const Duration(minutes: 5),
    this.autoSyncOnNetworkChange = true,
  });
}

核心代码示例

代码示例1：同步操作类

enum SyncStatus {
  pending,
  syncing,
  completed,
  failed,
  skipped,
}

enum SyncPriority {
  low,
  normal,
  high,
  urgent,
}

class SyncOperation {
  final String id;
  final String type;
  final String tableName;
  final Map<String, dynamic> data;
  final SyncStatus status;
  final SyncPriority priority;
  final DateTime createdAt;
  final DateTime? syncedAt;
  final int retryCount;
  final String? errorMessage;

  SyncOperation({
    required this.id,
    required this.type,
    required this.tableName,
    required this.data,
    this.status = SyncStatus.pending,
    this.priority = SyncPriority.normal,
    DateTime? createdAt,
    this.syncedAt,
    this.retryCount = 0,
    this.errorMessage,
  }) : createdAt = createdAt ?? DateTime.now();

  SyncOperation copyWith({
    SyncStatus? status,
    DateTime? syncedAt,
    int? retryCount,
    String? errorMessage,
  }) {
    return SyncOperation(
      id: id,
      type: type,
      tableName: tableName,
      data: data,
      status: status ?? this.status,
      priority: priority,
      createdAt: createdAt,
      syncedAt: syncedAt ?? this.syncedAt,
      retryCount: retryCount ?? this.retryCount,
      errorMessage: errorMessage ?? this.errorMessage,
    );
  }
}

代码说明：

1. 操作状态：定义5种同步状态，用于追踪操作进度
2. 操作优先级：定义4种优先级，用于控制同步顺序
3. 操作属性：包含操作类型、表名、数据、创建时间等信息
4. 复制方法：提供copyWith()方法，方便更新操作状态

代码示例2：同步队列

class SyncQueue {
  final List<SyncOperation> _operations = [];
  final int _maxRetries;

  SyncQueue({int maxRetries = 3}) : _maxRetries = maxRetries;

  void enqueue(SyncOperation operation) {
    _operations.add(operation);
    _sortByPriority();
  }

  SyncOperation? dequeue() {
    if (_operations.isEmpty) return null;
    
    final pendingOps = _operations.where((op) => op.status == SyncStatus.pending).toList();
    if (pendingOps.isEmpty) return null;

    pendingOps.sort((a, b) => a.createdAt.compareTo(b.createdAt));
    return pendingOps.first;
  }

  void remove(String id) {
    _operations.removeWhere((op) => op.id == id);
  }

  void update(SyncOperation operation) {
    final index = _operations.indexWhere((op) => op.id == operation.id);
    if (index != -1) {
      _operations[index] = operation;
    }
  }

  List<SyncOperation> getPendingOperations() {
    return _operations.where((op) => op.status == SyncStatus.pending).toList();
  }

  List<SyncOperation> getFailedOperations() {
    return _operations.where((op) => op.status == SyncStatus.failed).toList();
  }

  int getPendingCount() {
    return _operations.where((op) => op.status == SyncStatus.pending).length;
  }

  int getFailedCount() {
    return _operations.where((op) => op.status == SyncStatus.failed).length;
  }

  void clear() {
    _operations.clear();
  }

  void _sortByPriority() {
    _operations.sort((a, b) {
      final priorityOrder = [SyncPriority.urgent, SyncPriority.high, SyncPriority.normal, SyncPriority.low];
      return priorityOrder.indexOf(a.priority).compareTo(priorityOrder.indexOf(b.priority));
    });
  }
}

代码说明：

1. 入队操作：添加操作到队列，并按优先级排序
2. 出队操作：获取优先级最高、创建最早的待同步操作
3. 更新操作：更新队列中操作的状态
4. 查询操作：提供多种查询方法，获取待同步、失败的操作
5. 优先级排序：确保紧急操作优先执行

代码示例3：同步服务

class SyncService {
  final SyncQueue _queue;
  bool _isSyncing = false;
  final int _batchSize;

  SyncService({SyncQueue? queue, int batchSize = 10})
      : _queue = queue ?? SyncQueue(),
        _batchSize = batchSize;

  Future<void> startSync() async {
    if (_isSyncing) {
      return;
    }

    _isSyncing = true;

    try {
      while (_queue.getPendingCount() > 0) {
        final batch = _queue.getPendingOperations().take(_batchSize).toList();
        
        for (final operation in batch) {
          await _syncOperation(operation);
        }

        await Future.delayed(const Duration(seconds: 1));
      }
    } finally {
      _isSyncing = false;
    }
  }

  Future<void> _syncOperation(SyncOperation operation) async {
    _queue.update(operation.copyWith(status: SyncStatus.syncing));

    try {
      await _sendToServer(operation);
      
      _queue.update(operation.copyWith(
        status: SyncStatus.completed,
        syncedAt: DateTime.now(),
      ));
      
      _queue.remove(operation.id);
    } catch (e) {
      final newRetryCount = operation.retryCount + 1;
      
      if (newRetryCount >= _queue._maxRetries) {
        _queue.update(operation.copyWith(
          status: SyncStatus.failed,
          retryCount: newRetryCount,
          errorMessage: e.toString(),
        ));
      } else {
        _queue.update(operation.copyWith(
          status: SyncStatus.pending,
          retryCount: newRetryCount,
          errorMessage: e.toString(),
        ));
      }
    }
  }

  Future<void> _sendToServer(SyncOperation operation) async {
    await Future.delayed(const Duration(milliseconds: 200));
  }

  void enqueueOperation(String type, String tableName, Map<String, dynamic> data, {SyncPriority priority = SyncPriority.normal}) {
    final operation = SyncOperation(
      id: 'op_${DateTime.now().millisecondsSinceEpoch}',
      type: type,
      tableName: tableName,
      data: data,
      priority: priority,
    );
    _queue.enqueue(operation);
  }

  bool get isSyncing => _isSyncing;
  SyncQueue get queue => _queue;
}

代码说明：

1. 批量同步：支持批量处理同步操作，提高效率
2. 状态管理：自动更新操作状态，跟踪同步进度
3. 重试机制：失败后自动重试，达到最大重试次数后标记为失败
4. 并发控制：确保同一时间只有一个同步任务在执行
5. 优先级支持：支持按优先级处理同步操作

代码示例4：网络同步管理器

import 'package:connectivity_plus/connectivity_plus.dart';

class NetworkSyncManager {
  final SyncService _syncService;
  bool _isConnected = true;
  DateTime? _lastSyncTime;
  late Connectivity _connectivity;

  NetworkSyncManager({SyncService? syncService})
      : _syncService = syncService ?? SyncService() {
    _connectivity = Connectivity();
  }

  Future<void> startMonitoring() async {
    _connectivity.onConnectivityChanged.listen((result) {
      _isConnected = result != ConnectivityResult.none;
      if (_isConnected) {
        checkAndSync();
      }
    });
  }

  Future<void> stopMonitoring() async {
    await _connectivity.dispose();
  }

  Future<void> checkAndSync() async {
    if (!_isConnected) {
      return;
    }

    if (_syncService.isSyncing) {
      return;
    }

    await _syncService.startSync();
    _lastSyncTime = DateTime.now();
  }

  void setNetworkStatus(bool isConnected) {
    _isConnected = isConnected;
    if (isConnected) {
      checkAndSync();
    }
  }

  int getPendingCount() {
    return _syncService.queue.getPendingCount();
  }

  int getFailedCount() {
    return _syncService.queue.getFailedCount();
  }

  DateTime? get lastSyncTime => _lastSyncTime;
}

代码说明：

1. 网络监听：使用connectivity_plus监听网络状态变化
2. 自动同步：网络恢复后自动触发同步
3. 状态管理：跟踪网络连接状态和最后同步时间
4. 查询方法：提供待同步和失败操作的数量查询

高级特性

冲突检测与解决

class ConflictResolver {
  static SyncOperation resolveConflict(
    SyncOperation local,
    Map<String, dynamic> remote,
  ) {
    final localTime = local.createdAt;
    final remoteTime = DateTime.parse(remote['updatedAt'] as String);

    if (localTime.isAfter(remoteTime)) {
      return local;
    }

    return local.copyWith(data: remote);
  }
}

数据版本控制

class VersionControl {
  static int generateVersion() {
    return DateTime.now().millisecondsSinceEpoch;
  }

  static bool isVersionNewer(int localVersion, int remoteVersion) {
    return localVersion < remoteVersion;
  }
}

同步进度监听

class SyncProgressListener {
  static void onProgress(int current, int total) {
    final percentage = (current / total * 100).round();
    print('同步进度: $current/$total ($percentage%)');
  }
}

最佳实践

1. 本地数据优先

Future<void> saveUserData(Map<String, dynamic> data) async {
  await localDatabase.insert('users', data);
  
  _syncService.enqueueOperation(
    'insert',
    'users',
    data,
    priority: SyncPriority.high,
  );
}

2. 使用事务保证一致性

Future<void> syncWithTransaction() async {
  await database.transaction((txn) async {
    for (final op in pendingOperations) {
      await _applyOperation(txn, op);
      await _syncOperation(op);
    }
  });
}

3. 处理同步失败

Future<void> handleFailedOperations() async {
  final failedOps = _syncService.queue.getFailedOperations();
  
  for (final op in failedOps) {
    showNotification('同步失败', '${op.tableName}: ${op.errorMessage}');
  }
}

4. 定期同步

class PeriodicSync {
  static void startPeriodicSync(Duration interval) {
    Timer.periodic(interval, (_) {
      _syncManager.checkAndSync();
    });
  }
}

5. 网络质量评估

class NetworkQuality {
  static Future<bool> isNetworkStable() async {
    final result = await InternetAddress.lookup('example.com');
    return result.isNotEmpty && result[0].rawAddress.isNotEmpty;
  }
}

鸿蒙平台适配

网络状态监听

// 鸿蒙平台网络状态监听
final connectivity = Connectivity();
connectivity.onConnectivityChanged.listen((result) {
  switch (result) {
    case ConnectivityResult.wifi:
      print('已连接WiFi');
      break;
    case ConnectivityResult.mobile:
      print('已连接移动网络');
      break;
    case ConnectivityResult.none:
      print('网络已断开');
      break;
  }
});

平台差异

平台	网络检测	后台同步
Android	ConnectivityManager	WorkManager
iOS	Reachability	BackgroundTasks
HarmonyOS	NetManager	BackgroundTaskManager

权限配置

在鸿蒙平台上，需要在module.json5中配置权限：

{
  "module": {
    "abilities": [],
    "permission": [
      {
        "name": "ohos.permission.INTERNET"
      },
      {
        "name": "ohos.permission.GET_NETWORK_INFO"
      }
    ]
  }
}

常见问题

Q1: 网络频繁切换怎么办？

A：添加防抖机制，避免在网络不稳定时频繁触发同步。

Q2: 同步数据量很大怎么办？

A：使用分页和批量处理，避免一次性发送过多数据。

Q3: 如何处理数据冲突？

A：使用版本号或时间戳进行冲突检测，实现冲突解决策略。

Q4: 同步失败后如何恢复？

A：提供重试机制和手动同步功能，让用户可以手动触发同步。

Q5: 是否支持后台同步？

A：支持，使用平台提供的后台任务机制实现后台同步。

总结

离线数据同步是提升应用可靠性和用户体验的关键技术。通过本文的学习，你应该掌握了：

1. 核心概念：理解离线同步架构和同步状态
2. 同步操作：定义同步操作类，包含状态、优先级等属性
3. 同步队列：实现优先级队列，管理待同步操作
4. 同步服务：实现批量同步、重试机制和并发控制
5. 网络管理：监听网络状态变化，自动触发同步
6. 高级特性：冲突检测、版本控制、进度监听
7. 最佳实践：本地优先、事务保证、失败处理
8. 鸿蒙适配：了解鸿蒙平台的网络监听和权限配置

在实际项目中，合理实现离线数据同步可以确保用户在网络不稳定时也能正常使用应用，提升用户体验。