概述
在跨平台 Flutter 应用中，数据同步和状态管理是确保用户体验一致性的关键技术。本文将深入探讨跨平台数据同步策略、状态管理架构、冲突解决机制和性能优化技术。
数据同步架构
1. 分层同步架构
// 数据同步架构
abstract class DataSyncLayer {
// 同步数据
Future<SyncResult> syncData(SyncRequest request);
// 获取同步状态
Future<SyncStatus> getSyncStatus();
// 设置同步监听器
void setSyncListener(SyncListener listener);
// 清理资源
Future<void> dispose();
}
// 同步请求
class SyncRequest {
final String dataType;
final Map<String, dynamic>? parameters;
final DateTime? lastSyncTime;
final SyncMode mode;
SyncRequest({
required this.dataType,
this.parameters,
this.lastSyncTime,
this.mode = SyncMode.incremental,
});
}
// 同步模式
enum SyncMode {
full, // 全量同步
incremental, // 增量同步
biDirectional, // 双向同步
}
// 同步结果
class SyncResult {
final bool success;
final String? errorMessage;
final int syncedCount;
final int conflictCount;
final DateTime syncTime;
final Map<String, dynamic>? metadata;
SyncResult({
required this.success,
this.errorMessage,
required this.syncedCount,
required this.conflictCount,
required this.syncTime,
this.metadata,
});
}
// 同步状态
class SyncStatus {
final bool isSyncing;
final DateTime? lastSyncTime;
final DateTime? nextSyncTime;
final SyncProgress? progress;
final List<SyncError>? errors;
SyncStatus({
required this.isSyncing,
this.lastSyncTime,
this.nextSyncTime,
this.progress,
this.errors,
});
}
// 同步进度
class SyncProgress {
final int totalItems;
final int processedItems;
final String? currentItem;
SyncProgress({
required this.totalItems,
required this.processedItems,
this.currentItem,
});
double get percentage => totalItems > 0 ? processedItems / totalItems : 0.0;
}
// 同步错误
class SyncError {
final String code;
final String message;
final String? itemId;
final DateTime timestamp;
SyncError({
required this.code,
required this.message,
this.itemId,
DateTime? timestamp,
}) : timestamp = timestamp ?? DateTime.now();
}
// 同步监听器
abstract class SyncListener {
void onSyncStarted(String dataType);
void onSyncProgress(String dataType, SyncProgress progress);
void onSyncCompleted(String dataType, SyncResult result);
void onSyncError(String dataType, SyncError error);
void onConflictDetected(String dataType, List<DataConflict> conflicts);
}2. 平台特定同步实现
// 移动平台同步实现
class MobileDataSync extends DataSyncLayer {
static const MethodChannel _channel = MethodChannel('mobile_data_sync');
final List<SyncListener> _listeners = [];
SyncStatus _currentStatus = SyncStatus(isSyncing: false);
@override
Future<SyncResult> syncData(SyncRequest request) async {
try {
_updateStatus(SyncStatus(isSyncing: true));
_notifySyncStarted(request.dataType);
final result = await _channel.invokeMethod<Map<String, dynamic>>('syncData', {
'dataType': request.dataType,
'parameters': request.parameters,
'lastSyncTime': request.lastSyncTime?.millisecondsSinceEpoch,
'mode': request.mode.index,
});
final syncResult = SyncResult(
success: result?['success'] as bool? ?? false,
errorMessage: result?['errorMessage'] as String?,
syncedCount: result?['syncedCount'] as int? ?? 0,
conflictCount: result?['conflictCount'] as int? ?? 0,
syncTime: DateTime.fromMillisecondsSinceEpoch(
result?['syncTime'] as int? ?? DateTime.now().millisecondsSinceEpoch,
),
metadata: result?['metadata'] as Map<String, dynamic>?,
);
_updateStatus(SyncStatus(isSyncing: false, lastSyncTime: syncResult.syncTime));
_notifySyncCompleted(request.dataType, syncResult);
return syncResult;
} on PlatformException catch (e) {
final error = SyncError(
code: e.code,
message: e.message ?? 'Unknown sync error',
);
_updateStatus(SyncStatus(
isSyncing: false,
errors: [error],
));
_notifySyncError(request.dataType, error);
return SyncResult(
success: false,
errorMessage: e.message,
syncedCount: 0,
conflictCount: 0,
syncTime: DateTime.now(),
);
}
}
@override
Future<SyncStatus> getSyncStatus() async {
try {
final result = await _channel.invokeMethod<Map<String, dynamic>>('getSyncStatus');
_currentStatus = SyncStatus(
isSyncing: result?['isSyncing'] as bool? ?? false,
lastSyncTime: result?['lastSyncTime'] != null
? DateTime.fromMillisecondsSinceEpoch(result!['lastSyncTime'] as int)
: null,
nextSyncTime: result?['nextSyncTime'] != null
? DateTime.fromMillisecondsSinceEpoch(result!['nextSyncTime'] as int)
: null,
progress: result?['progress'] != null
? SyncProgress(
totalItems: result!['progress']['totalItems'] as int,
processedItems: result['progress']['processedItems'] as int,
currentItem: result['progress']['currentItem'] as String?,
)
: null,
);
return _currentStatus;
} on PlatformException catch (e) {
return _currentStatus;
}
}
@override
void setSyncListener(SyncListener listener) {
_listeners.add(listener);
}
@override
Future<void> dispose() async {
_listeners.clear();
await _channel.invokeMethod('dispose');
}
void _updateStatus(SyncStatus status) {
_currentStatus = status;
}
void _notifySyncStarted(String dataType) {
for (final listener in _listeners) {
listener.onSyncStarted(dataType);
}
}
void _notifySyncCompleted(String dataType, SyncResult result) {
for (final listener in _listeners) {
listener.onSyncCompleted(dataType, result);
}
}
void _notifySyncError(String dataType, SyncError error) {
for (final listener in _listeners) {
listener.onSyncError(dataType, error);
}
}
void _notifyConflictDetected(String dataType, List<DataConflict> conflicts) {
for (final listener in _listeners) {
listener.onConflictDetected(dataType, conflicts);
}
}
}
// Web平台同步实现
class WebDataSync extends DataSyncLayer {
final List<SyncListener> _listeners = [];
SyncStatus _currentStatus = SyncStatus(isSyncing: false);
@override
Future<SyncResult> syncData(SyncRequest request) async {
try {
_updateStatus(SyncStatus(isSyncing: true));
_notifySyncStarted(request.dataType);
// 使用IndexedDB或Web Storage进行同步
final result = await _performWebSync(request);
_updateStatus(SyncStatus(isSyncing: false, lastSyncTime: result.syncTime));
_notifySyncCompleted(request.dataType, result);
return result;
} catch (e) {
final error = SyncError(
code: 'WEB_SYNC_ERROR',
message: e.toString(),
);
_updateStatus(SyncStatus(
isSyncing: false,
errors: [error],
));
_notifySyncError(request.dataType, error);
return SyncResult(
success: false,
errorMessage: e.toString(),
syncedCount: 0,
conflictCount: 0,
syncTime: DateTime.now(),
);
}
}
Future<SyncResult> _performWebSync(SyncRequest request) async {
// 实现Web平台特定的同步逻辑
// 可以使用Service Worker、IndexedDB等
// 模拟实现
await Future.delayed(Duration(seconds: 2));
return SyncResult(
success: true,
syncedCount: 10,
conflictCount: 0,
syncTime: DateTime.now(),
);
}
@override
Future<SyncStatus> getSyncStatus() async {
return _currentStatus;
}
@override
void setSyncListener(SyncListener listener) {
_listeners.add(listener);
}
@override
Future<void> dispose() async {
_listeners.clear();
}
void _updateStatus(SyncStatus status) {
_currentStatus = status;
}
void _notifySyncStarted(String dataType) {
for (final listener in _listeners) {
listener.onSyncStarted(dataType);
}
}
void _notifySyncCompleted(String dataType, SyncResult result) {
for (final listener in _listeners) {
listener.onSyncCompleted(dataType, result);
}
}
void _notifySyncError(String dataType, SyncError error) {
for (final listener in _listeners) {
listener.onSyncError(dataType, error);
}
}
void _notifyConflictDetected(String dataType, List<DataConflict> conflicts) {
for (final listener in _listeners) {
listener.onConflictDetected(dataType, conflicts);
}
}
}状态管理架构
1. 跨平台状态管理器
// 跨平台状态管理器
abstract class CrossPlatformStateManager {
// 获取状态
T? getState<T>(String key);
// 设置状态
Future<void> setState<T>(String key, T value);
// 监听状态变化
Stream<T> getStateStream<T>(String key);
// 清除状态
Future<void> clearState(String key);
// 清除所有状态
Future<void> clearAllStates();
// 持久化状态
Future<void> persistState();
// 恢复状态
Future<void> restoreState();
}
// 状态变化事件
class StateChangeEvent<T> {
final String key;
final T? oldValue;
final T newValue;
final DateTime timestamp;
StateChangeEvent({
required this.key,
this.oldValue,
required this.newValue,
DateTime? timestamp,
}) : timestamp = timestamp ?? DateTime.now();
}
// 移动平台状态管理实现
class MobileStateManager extends CrossPlatformStateManager {
static const MethodChannel _channel = MethodChannel('mobile_state_manager');
final Map<String, StreamController> _controllers = {};
final Map<String, dynamic> _stateCache = {};
@override
T? getState<T>(String key) {
// 先从缓存获取
if (_stateCache.containsKey(key)) {
return _stateCache[key] as T?;
}
// 从原生平台获取
try {
final result = _channel.invokeMethod<T>('getState', {'key': key});
if (result != null) {
_stateCache[key] = result;
}
return result;
} on PlatformException catch (e) {
print('Error getting state for key $key: ${e.message}');
return null;
}
}
@override
Future<void> setState<T>(String key, T value) async {
try {
final oldValue = _stateCache[key];
// 更新缓存
_stateCache[key] = value;
// 更新原生平台状态
await _channel.invokeMethod('setState', {
'key': key,
'value': value,
});
// 通知监听器
_notifyStateChange(key, oldValue, value);
} on PlatformException catch (e) {
print('Error setting state for key $key: ${e.message}');
// 回滚缓存
_stateCache.remove(key);
rethrow;
}
}
@override
Stream<T> getStateStream<T>(String key) {
final streamKey = '${key}_${T.toString()}';
if (!_controllers.containsKey(streamKey)) {
final controller = StreamController<T>.broadcast();
_controllers[streamKey] = controller;
// 设置原生平台监听
_channel.setMethodCallHandler((call) async {
if (call.method == 'onStateChanged') {
final arguments = call.arguments as Map<String, dynamic>;
final changedKey = arguments['key'] as String;
final newValue = arguments['value'] as T?;
if (changedKey == key) {
_stateCache[key] = newValue;
controller.add(newValue);
}
}
});
// 通知原生平台开始监听
_channel.invokeMethod('startListening', {'key': key});
}
return _controllers[streamKey]!.stream.cast<T>();
}
@override
Future<void> clearState(String key) async {
try {
_stateCache.remove(key);
await _channel.invokeMethod('clearState', {'key': key});
_notifyStateChange(key, _stateCache[key], null);
} on PlatformException catch (e) {
print('Error clearing state for key $key: ${e.message}');
}
}
@override
Future<void> clearAllStates() async {
try {
_stateCache.clear();
await _channel.invokeMethod('clearAllStates');
// 通知所有监听器
for (final controller in _controllers.values) {
controller.add(null);
}
} on PlatformException catch (e) {
print('Error clearing all states: ${e.message}');
}
}
@override
Future<void> persistState() async {
try {
await _channel.invokeMethod('persistState');
} on PlatformException catch (e) {
print('Error persisting state: ${e.message}');
}
}
@override
Future<void> restoreState() async {
try {
final result = await _channel.invokeMethod<Map<String, dynamic>>('restoreState');
if (result != null) {
_stateCache.addAll(result);
}
} on PlatformException catch (e) {
print('Error restoring state: ${e.message}');
}
}
void _notifyStateChange<T>(String key, T? oldValue, T? newValue) {
final streamKey = '${key}_${T.toString()}';
final controller = _controllers[streamKey];
if (controller != null && !controller.isClosed) {
controller.add(newValue);
}
}
@override
Future<void> dispose() async {
for (final controller in _controllers.values) {
controller.close();
}
_controllers.clear();
_stateCache.clear();
await _channel.invokeMethod('dispose');
}
}
// Web平台状态管理实现
class WebStateManager extends CrossPlatformStateManager {
final Map<String, StreamController> _controllers = {};
final Map<String, dynamic> _stateCache = {};
final Storage _storage = html.window.localStorage;
@override
T? getState<T>(String key) {
// 先从缓存获取
if (_stateCache.containsKey(key)) {
return _stateCache[key] as T?;
}
// 从localStorage获取
try {
final value = _storage[key];
if (value != null) {
final decoded = json.decode(value);
_stateCache[key] = decoded;
return decoded as T?;
}
} catch (e) {
print('Error getting state for key $key: $e');
}
return null;
}
@override
Future<void> setState<T>(String key, T value) async {
try {
final oldValue = _stateCache[key];
// 更新缓存
_stateCache[key] = value;
// 更新localStorage
_storage[key] = json.encode(value);
// 通知监听器
_notifyStateChange(key, oldValue, value);
} catch (e) {
print('Error setting state for key $key: $e');
// 回滚缓存
_stateCache.remove(key);
rethrow;
}
}
@override
Stream<T> getStateStream<T>(String key) {
final streamKey = '${key}_${T.toString()}';
if (!_controllers.containsKey(streamKey)) {
final controller = StreamController<T>.broadcast();
_controllers[streamKey] = controller;
// 监听storage事件
html.window.addEventListener('storage', (event) {
final storageEvent = event as html.StorageEvent;
if (storageEvent.key == key) {
final newValue = storageEvent.newValue;
T? decodedValue;
if (newValue != null) {
try {
decodedValue = json.decode(newValue) as T?;
} catch (e) {
print('Error decoding stored value: $e');
}
}
_stateCache[key] = decodedValue;
controller.add(decodedValue);
}
});
}
return _controllers[streamKey]!.stream.cast<T>();
}
@override
Future<void> clearState(String key) async {
try {
_stateCache.remove(key);
_storage.remove(key);
_notifyStateChange(key, _stateCache[key], null);
} catch (e) {
print('Error clearing state for key $key: $e');
}
}
@override
Future<void> clearAllStates() async {
try {
_stateCache.clear();
_storage.clear();
// 通知所有监听器
for (final controller in _controllers.values) {
controller.add(null);
}
} catch (e) {
print('Error clearing all states: $e');
}
}
@override
Future<void> persistState() async {
// Web平台的localStorage本身就是持久化的
// 这里可以添加额外的持久化逻辑
}
@override
Future<void> restoreState() async {
// Web平台的localStorage数据会自动恢复
// 这里可以添加额外的恢复逻辑
}
void _notifyStateChange<T>(String key, T? oldValue, T? newValue) {
final streamKey = '${key}_${T.toString()}';
final controller = _controllers[streamKey];
if (controller != null && !controller.isClosed) {
controller.add(newValue);
}
}
@override
Future<void> dispose() async {
for (final controller in _controllers.values) {
controller.close();
}
_controllers.clear();
_stateCache.clear();
}
}冲突解决机制
1. 数据冲突检测
// 数据冲突
class DataConflict {
final String itemId;
final String dataType;
final dynamic localValue;
final dynamic remoteValue;
final DateTime localTimestamp;
final DateTime remoteTimestamp;
final ConflictType type;
DataConflict({
required this.itemId,
required this.dataType,
required this.localValue,
required this.remoteValue,
required this.localTimestamp,
required this.remoteTimestamp,
required this.type,
});
}
// 冲突类型
enum ConflictType {
update, // 更新冲突
delete, // 删除冲突
create, // 创建冲突
}
// 冲突解决策略
enum ConflictResolutionStrategy {
useLocal, // 使用本地数据
useRemote, // 使用远程数据
merge, // 合并数据
manual, // 手动解决
timestamp, // 基于时间戳
custom, // 自定义策略
}
// 冲突解决器
abstract class ConflictResolver {
Future<ConflictResolution> resolve(DataConflict conflict);
}
// 冲突解决结果
class ConflictResolution {
final ConflictResolutionStrategy strategy;
final dynamic resolvedValue;
final String? reason;
ConflictResolution({
required this.strategy,
required this.resolvedValue,
this.reason,
});
}
// 时间戳冲突解决器
class TimestampConflictResolver implements ConflictResolver {
@override
Future<ConflictResolution> resolve(DataConflict conflict) async {
if (conflict.localTimestamp.isAfter(conflict.remoteTimestamp)) {
return ConflictResolution(
strategy: ConflictResolutionStrategy.timestamp,
resolvedValue: conflict.localValue,
reason: 'Local data is newer',
);
} else {
return ConflictResolution(
strategy: ConflictResolutionStrategy.timestamp,
resolvedValue: conflict.remoteValue,
reason: 'Remote data is newer',
);
}
}
}
// 合并冲突解决器
class MergeConflictResolver implements ConflictResolver {
@override
Future<ConflictResolution> resolve(DataConflict conflict) async {
try {
final mergedValue = await _mergeValues(
conflict.localValue,
conflict.remoteValue,
conflict.dataType,
);
return ConflictResolution(
strategy: ConflictResolutionStrategy.merge,
resolvedValue: mergedValue,
reason: 'Data merged successfully',
);
} catch (e) {
// 合并失败，回退到时间戳策略
final timestampResolver = TimestampConflictResolver();
return await timestampResolver.resolve(conflict);
}
}
Future<dynamic> _mergeValues(dynamic localValue, dynamic remoteValue, String dataType) async {
switch (dataType) {
case 'map':
return _mergeMaps(localValue as Map, remoteValue as Map);
case 'list':
return _mergeLists(localValue as List, remoteValue as List);
case 'string':
return _mergeStrings(localValue as String, remoteValue as String);
default:
throw ArgumentError('Unsupported data type for merging: $dataType');
}
}
Map<String, dynamic> _mergeMaps(Map<String, dynamic> local, Map<String, dynamic> remote) {
final merged = Map<String, dynamic>.from(local);
merged.addAll(remote);
return merged;
}
List<dynamic> _mergeLists(List<dynamic> local, List<dynamic> remote) {
final merged = <dynamic>[];
merged.addAll(local);
for (final item in remote) {
if (!merged.contains(item)) {
merged.add(item);
}
}
return merged;
}
String _mergeStrings(String local, String remote) {
// 简单的字符串合并策略
return '$local\n$remote';
}
}
// 自定义冲突解决器
class CustomConflictResolver implements ConflictResolver {
final ConflictResolutionFunction _resolveFunction;
CustomConflictResolver(this._resolveFunction);
@override
Future<ConflictResolution> resolve(DataConflict conflict) async {
return await _resolveFunction(conflict);
}
}
// 冲突解决函数类型定义
typedef ConflictResolutionFunction = Future<ConflictResolution> Function(DataConflict conflict);
// 冲突管理器
class ConflictManager {
final Map<String, ConflictResolver> _resolvers = {};
final List<ConflictListener> _listeners = [];
ConflictResolutionStrategy _defaultStrategy = ConflictResolutionStrategy.timestamp;
// 注册冲突解决器
void registerResolver(String dataType, ConflictResolver resolver) {
_resolvers[dataType] = resolver;
}
// 设置默认解决策略
void setDefaultStrategy(ConflictResolutionStrategy strategy) {
_defaultStrategy = strategy;
}
// 添加冲突监听器
void addConflictListener(ConflictListener listener) {
_listeners.add(listener);
}
// 移除冲突监听器
void removeConflictListener(ConflictListener listener) {
_listeners.remove(listener);
}
// 解决冲突
Future<ConflictResolution> resolveConflict(DataConflict conflict) async {
// 通知监听器
for (final listener in _listeners) {
listener.onConflictDetected(conflict);
}
// 获取解决器
final resolver = _resolvers[conflict.dataType];
if (resolver != null) {
return await resolver.resolve(conflict);
}
// 使用默认策略
return await _resolveWithDefaultStrategy(conflict);
}
// 批量解决冲突
Future<List<ConflictResolution>> resolveConflicts(List<DataConflict> conflicts) async {
final resolutions = <ConflictResolution>[];
for (final conflict in conflicts) {
try {
final resolution = await resolveConflict(conflict);
resolutions.add(resolution);
} catch (e) {
// 解决失败，使用默认策略
final resolution = await _resolveWithDefaultStrategy(conflict);
resolutions.add(resolution);
}
}
return resolutions;
}
// 使用默认策略解决冲突
Future<ConflictResolution> _resolveWithDefaultStrategy(DataConflict conflict) async {
switch (_defaultStrategy) {
case ConflictResolutionStrategy.useLocal:
return ConflictResolution(
strategy: ConflictResolutionStrategy.useLocal,
resolvedValue: conflict.localValue,
reason: 'Using default strategy: use local',
);
case ConflictResolutionStrategy.useRemote:
return ConflictResolution(
strategy: ConflictResolutionStrategy.useRemote,
resolvedValue: conflict.remoteValue,
reason: 'Using default strategy: use remote',
);
case ConflictResolutionStrategy.timestamp:
final timestampResolver = TimestampConflictResolver();
return await timestampResolver.resolve(conflict);
case ConflictResolutionStrategy.manual:
return ConflictResolution(
strategy: ConflictResolutionStrategy.manual,
resolvedValue: conflict.localValue, // 需要用户手动选择
reason: 'Manual resolution required',
);
default:
throw ArgumentError('Unsupported default strategy: $_defaultStrategy');
}
}
}
// 冲突监听器
abstract class ConflictListener {
void onConflictDetected(DataConflict conflict);
void onConflictResolved(DataConflict conflict, ConflictResolution resolution);
}2. 自动冲突解决
// 自动冲突解决管理器
class AutoConflictResolver {
final ConflictManager _conflictManager;
final Map<String, AutoResolutionRule> _rules = {};
AutoConflictResolver(this._conflictManager) {
_setupDefaultRules();
}
// 设置自动解决规则
void setAutoResolutionRule(String dataType, AutoResolutionRule rule) {
_rules[dataType] = rule;
}
// 自动解决冲突
Future<List<ConflictResolution>> autoResolveConflicts(List<DataConflict> conflicts) async {
final resolutions = <ConflictResolution>[];
final manualConflicts = <DataConflict>[];
for (final conflict in conflicts) {
final rule = _rules[conflict.dataType];
if (rule != null && rule.shouldAutoResolve(conflict)) {
try {
final resolution = await rule.resolve(conflict);
resolutions.add(resolution);
} catch (e) {
// 自动解决失败，添加到手动解决列表
manualConflicts.add(conflict);
}
} else {
// 需要手动解决
manualConflicts.add(conflict);
}
}
// 处理需要手动解决的冲突
if (manualConflicts.isNotEmpty) {
final manualResolutions = await _conflictManager.resolveConflicts(manualConflicts);
resolutions.addAll(manualResolutions);
}
return resolutions;
}
// 设置默认规则
void _setupDefaultRules() {
// 用户数据优先使用最新时间戳
setAutoResolutionRule('user_profile', TimestampAutoResolutionRule());
// 配置数据优先使用远程
setAutoResolutionRule('app_config', RemotePriorityAutoResolutionRule());
// 日志数据合并
setAutoResolutionRule('logs', MergeAutoResolutionRule());
}
}
// 自动解决规则
abstract class AutoResolutionRule {
bool shouldAutoResolve(DataConflict conflict);
Future<ConflictResolution> resolve(DataConflict conflict);
}
// 时间戳自动解决规则
class TimestampAutoResolutionRule implements AutoResolutionRule {
@override
bool shouldAutoResolve(DataConflict conflict) {
return true; // 总是自动解决
}
@override
Future<ConflictResolution> resolve(DataConflict conflict) async {
final resolver = TimestampConflictResolver();
return await resolver.resolve(conflict);
}
}
// 远程优先自动解决规则
class RemotePriorityAutoResolutionRule implements AutoResolutionRule {
@override
bool shouldAutoResolve(DataConflict conflict) {
return true; // 总是自动解决
}
@override
Future<ConflictResolution> resolve(DataConflict conflict) async {
return ConflictResolution(
strategy: ConflictResolutionStrategy.useRemote,
resolvedValue: conflict.remoteValue,
reason: 'Remote data has priority',
);
}
}
// 合并自动解决规则
class MergeAutoResolutionRule implements AutoResolutionRule {
@override
bool shouldAutoResolve(DataConflict conflict) {
return true; // 总是尝试合并
}
@override
Future<ConflictResolution> resolve(DataConflict conflict) async {
final resolver = MergeConflictResolver();
return await resolver.resolve(conflict);
}
}性能优化
1. 增量同步优化
// 增量同步管理器
class IncrementalSyncManager {
final DataSyncLayer _syncLayer;
final Map<String, DateTime> _lastSyncTimes = {};
final Map<String, String> _syncTokens = {};
IncrementalSyncManager(this._syncLayer);
// 执行增量同步
Future<SyncResult> performIncrementalSync(String dataType) async {
final lastSyncTime = _lastSyncTimes[dataType];
final syncToken = _syncTokens[dataType];
final request = SyncRequest(
dataType: dataType,
lastSyncTime: lastSyncTime,
mode: SyncMode.incremental,
parameters: syncToken != null ? {'syncToken': syncToken} : null,
);
final result = await _syncLayer.syncData(request);
if (result.success) {
// 更新同步时间和令牌
_lastSyncTimes[dataType] = result.syncTime;
if (result.metadata != null && result.metadata!.containsKey('syncToken')) {
_syncTokens[dataType] = result.metadata!['syncToken'] as String;
}
}
return result;
}
// 执行全量同步
Future<SyncResult> performFullSync(String dataType) async {
final request = SyncRequest(
dataType: dataType,
mode: SyncMode.full,
);
final result = await _syncLayer.syncData(request);
if (result.success) {
// 重置增量同步状态
_lastSyncTimes[dataType] = result.syncTime;
_syncTokens.remove(dataType);
}
return result;
}
// 智能同步（根据情况选择增量或全量）
Future<SyncResult> performSmartSync(String dataType) async {
final lastSyncTime = _lastSyncTimes[dataType];
// 如果从未同步或上次同步时间太久，执行全量同步
if (lastSyncTime == null ||
DateTime.now().difference(lastSyncTime).inDays > 7) {
return await performFullSync(dataType);
}
// 否则执行增量同步
return await performIncrementalSync(dataType);
}
// 获取同步状态
Map<String, dynamic> getSyncStatus() {
return {
'lastSyncTimes': _lastSyncTimes.map((key, value) => MapEntry(key, value.toIso8601String())),
'syncTokens': _syncTokens,
};
}
// 清理同步状态
void clearSyncStatus(String dataType) {
_lastSyncTimes.remove(dataType);
_syncTokens.remove(dataType);
}
// 清理所有同步状态
void clearAllSyncStatus() {
_lastSyncTimes.clear();
_syncTokens.clear();
}
}2. 批量操作优化
// 批量同步管理器
class BatchSyncManager {
final DataSyncLayer _syncLayer;
final Map<String, List<SyncRequest>> _pendingBatches = {};
final Map<String, Timer> _batchTimers = {};
final Duration _batchDelay;
BatchSyncManager(this._syncLayer, {Duration batchDelay = const Duration(seconds: 5)})
: _batchDelay = batchDelay;
// 添加同步请求到批处理
Future<SyncResult> addSyncRequest(SyncRequest request) async {
final dataType = request.dataType;
final batch = _pendingBatches.putIfAbsent(dataType, () => []);
batch.add(request);
// 设置批处理定时器
_scheduleBatchExecution(dataType);
// 等待批处理完成
return await _waitForBatchCompletion(dataType, request);
}
// 调度批处理执行
void _scheduleBatchExecution(String dataType) {
final timer = _batchTimers[dataType];
if (timer?.isActive == true) {
return; // 已有定时器在运行
}
_batchTimers[dataType] = Timer(_batchDelay, () {
_executeBatch(dataType);
});
}
// 执行批处理
Future<void> _executeBatch(String dataType) async {
final batch = _pendingBatches.remove(dataType);
_batchTimers.remove(dataType);
if (batch == null || batch.isEmpty) return;
try {
// 合并批处理请求
final mergedRequest = _mergeBatchRequests(batch);
// 执行同步
final result = await _syncLayer.syncData(mergedRequest);
// 通知所有等待的请求
_notifyBatchCompletion(dataType, result);
} catch (e) {
// 通知所有等待的请求失败
_notifyBatchError(dataType, e);
}
}
// 合并批处理请求
SyncRequest _mergeBatchRequests(List<SyncRequest> requests) {
if (requests.isEmpty) {
throw ArgumentError('Cannot merge empty batch');
}
final firstRequest = requests.first;
final earliestTime = requests
.map((r) => r.lastSyncTime)
.where((t) => t != null)
.reduce((a, b) => a!.isBefore(b!) ? a : b);
return SyncRequest(
dataType: firstRequest.dataType,
lastSyncTime: earliestTime,
mode: SyncMode.incremental,
parameters: {
'batchSize': requests.length,
'batchIds': requests.map((r) => r.parameters?['id']).toList(),
},
);
}
// 等待批处理完成
Future<SyncResult> _waitForBatchCompletion(String dataType, SyncRequest request) async {
final completer = Completer<SyncResult>();
// 注册完成监听器
final listener = _BatchCompletionListener(
dataType: dataType,
requestId: request.parameters?['id'],
completer: completer,
);
_addCompletionListener(listener);
return await completer.future;
}
// 通知批处理完成
void _notifyBatchCompletion(String dataType, SyncResult result) {
final listeners = _getCompletionListeners(dataType);
for (final listener in listeners) {
listener.completer.complete(result);
}
_removeCompletionListeners(dataType);
}
// 通知批处理错误
void _notifyBatchError(String dataType, dynamic error) {
final listeners = _getCompletionListeners(dataType);
for (final listener in listeners) {
listener.completer.completeError(error);
}
_removeCompletionListeners(dataType);
}
// 批处理完成监听器管理
final Map<String, List<_BatchCompletionListener>> _completionListeners = {};
void _addCompletionListener(_BatchCompletionListener listener) {
final listeners = _completionListeners.putIfAbsent(listener.dataType, () => []);
listeners.add(listener);
}
List<_BatchCompletionListener> _getCompletionListeners(String dataType) {
return _completionListeners[dataType] ?? [];
}
void _removeCompletionListeners(String dataType) {
_completionListeners.remove(dataType);
}
// 强制执行所有待处理的批处理
Future<void> flushAllBatches() async {
for (final dataType in _pendingBatches.keys.toList()) {
_executeBatch(dataType);
}
}
}
// 批处理完成监听器
class _BatchCompletionListener {
final String dataType;
final String? requestId;
final Completer<SyncResult> completer;
_BatchCompletionListener({
required this.dataType,
this.requestId,
required this.completer,
});
}最佳实践
1. 架构设计
- 分层架构：将数据同步和状态管理分为不同层次
- 平台抽象：创建统一的平台抽象层
- 模块化设计：按功能模块组织代码
- 依赖注入：使用依赖注入管理组件生命周期
2. 性能优化
- 增量同步：只同步变化的数据
- 批量操作：合并多个操作减少网络请求
- 缓存策略：合理使用缓存减少同步频率
- 压缩传输：压缩数据减少传输时间
3. 错误处理
- 冲突检测：及时发现数据冲突
- 自动解决：使用智能策略自动解决冲突
- 手动干预：提供手动解决冲突的界面
- 重试机制：建立完善的重试机制
4. 安全考虑
- 数据加密：对敏感数据进行加密传输
- 身份认证：确保同步操作的安全性
- 访问控制：限制数据的访问权限
- 审计日志：记录所有同步操作
总结
Flutter 跨平台数据同步与状态管理桥接技术为构建一致性的跨平台应用提供了强大的支持。通过掌握数据同步架构、状态管理、冲突解决和性能优化等技术，开发者可以构建出稳定、高效的跨平台应用。
关键成功因素：
- 合理的架构设计
- 高效的同步策略
- 智能的冲突解决
- 优秀的性能优化
- 完善的错误处理
通过本文的学习，开发者应该能够掌握 Flutter 跨平台数据同步与状态管理的核心技术，构建出功能强大、用户体验一致的跨平台应用。