|
|
"""
|
|
|
最终一致性模式 - 数据同步系统
|
|
|
|
|
|
此模块实现了最终一致性模式，用于处理分布式系统中的数据同步问题。
|
|
|
最终一致性系统允许数据在一段时间内处于不一致状态，但保证最终会达到一致。
|
|
|
"""
|
|
|
|
|
|
from abc import ABC, abstractmethod
|
|
|
from enum import Enum, auto
|
|
|
from typing import Dict, List, Optional, Any, Set, Callable
|
|
|
import threading
|
|
|
import time
|
|
|
import uuid
|
|
|
import queue
|
|
|
import random
|
|
|
|
|
|
|
|
|
class SyncStatus(Enum):
|
|
|
"""同步状态枚举"""
|
|
|
PENDING = auto() # 待同步
|
|
|
SYNCHRONIZING = auto() # 正在同步
|
|
|
COMPLETED = auto() # 同步完成
|
|
|
FAILED = auto() # 同步失败
|
|
|
RETRYING = auto() # 重试中
|
|
|
|
|
|
|
|
|
class ReplicationMessage:
|
|
|
"""
|
|
|
复制消息类
|
|
|
|
|
|
表示需要在系统节点间复制的数据变更消息
|
|
|
"""
|
|
|
|
|
|
def __init__(self, operation: str, entity_id: str, data: Dict[str, Any], source_node: str):
|
|
|
"""
|
|
|
初始化复制消息
|
|
|
|
|
|
Args:
|
|
|
operation: 操作类型 (create, update, delete)
|
|
|
entity_id: 实体ID
|
|
|
data: 实体数据
|
|
|
source_node: 源节点ID
|
|
|
"""
|
|
|
self.message_id = str(uuid.uuid4())
|
|
|
self.operation = operation
|
|
|
self.entity_id = entity_id
|
|
|
self.data = data
|
|
|
self.source_node = source_node
|
|
|
self.timestamp = time.time()
|
|
|
self.version = 1
|
|
|
self.destination_nodes: Set[str] = set()
|
|
|
self.delivered_nodes: Set[str] = set()
|
|
|
self.failed_nodes: Set[str] = set()
|
|
|
|
|
|
def mark_delivered(self, node_id: str) -> None:
|
|
|
"""标记消息已送达指定节点"""
|
|
|
self.delivered_nodes.add(node_id)
|
|
|
if node_id in self.failed_nodes:
|
|
|
self.failed_nodes.remove(node_id)
|
|
|
|
|
|
def mark_failed(self, node_id: str) -> None:
|
|
|
"""标记消息送达指定节点失败"""
|
|
|
self.failed_nodes.add(node_id)
|
|
|
|
|
|
def is_delivered_to_all(self) -> bool:
|
|
|
"""检查消息是否已送达所有目标节点"""
|
|
|
return self.destination_nodes.issubset(self.delivered_nodes)
|
|
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
|
"""转换为字典表示"""
|
|
|
return {
|
|
|
"message_id": self.message_id,
|
|
|
"operation": self.operation,
|
|
|
"entity_id": self.entity_id,
|
|
|
"data": self.data,
|
|
|
"source_node": self.source_node,
|
|
|
"timestamp": self.timestamp,
|
|
|
"version": self.version,
|
|
|
"destination_nodes": list(self.destination_nodes),
|
|
|
"delivered_nodes": list(self.delivered_nodes),
|
|
|
"failed_nodes": list(self.failed_nodes)
|
|
|
}
|
|
|
|
|
|
|
|
|
class MessageQueue:
|
|
|
"""
|
|
|
消息队列类
|
|
|
|
|
|
用于在节点间传递复制消息
|
|
|
"""
|
|
|
|
|
|
def __init__(self, node_id: str):
|
|
|
"""
|
|
|
初始化消息队列
|
|
|
|
|
|
Args:
|
|
|
node_id: 节点ID
|
|
|
"""
|
|
|
self.node_id = node_id
|
|
|
self._queue = queue.Queue()
|
|
|
self._lock = threading.RLock()
|
|
|
|
|
|
def enqueue(self, message: ReplicationMessage) -> None:
|
|
|
"""
|
|
|
将消息加入队列
|
|
|
|
|
|
Args:
|
|
|
message: 复制消息
|
|
|
"""
|
|
|
with self._lock:
|
|
|
self._queue.put(message)
|
|
|
|
|
|
def dequeue(self, timeout: Optional[float] = None) -> Optional[ReplicationMessage]:
|
|
|
"""
|
|
|
从队列中取出消息
|
|
|
|
|
|
Args:
|
|
|
timeout: 超时时间
|
|
|
|
|
|
Returns:
|
|
|
消息，如果队列为空且超时则返回None
|
|
|
"""
|
|
|
try:
|
|
|
return self._queue.get(timeout=timeout)
|
|
|
except queue.Empty:
|
|
|
return None
|
|
|
|
|
|
def size(self) -> int:
|
|
|
"""
|
|
|
获取队列大小
|
|
|
|
|
|
Returns:
|
|
|
队列中的消息数量
|
|
|
"""
|
|
|
return self._queue.qsize()
|
|
|
|
|
|
|
|
|
class StorageProvider(ABC):
|
|
|
"""
|
|
|
存储提供者抽象基类
|
|
|
|
|
|
定义数据存储的接口
|
|
|
"""
|
|
|
|
|
|
@abstractmethod
|
|
|
def get(self, entity_id: str) -> Optional[Dict[str, Any]]:
|
|
|
"""获取实体"""
|
|
|
pass
|
|
|
|
|
|
@abstractmethod
|
|
|
def put(self, entity_id: str, data: Dict[str, Any]) -> None:
|
|
|
"""存储实体"""
|
|
|
pass
|
|
|
|
|
|
@abstractmethod
|
|
|
def delete(self, entity_id: str) -> bool:
|
|
|
"""删除实体"""
|
|
|
pass
|
|
|
|
|
|
@abstractmethod
|
|
|
def get_all(self) -> Dict[str, Dict[str, Any]]:
|
|
|
"""获取所有实体"""
|
|
|
pass
|
|
|
|
|
|
|
|
|
class InMemoryStorage(StorageProvider):
|
|
|
"""
|
|
|
内存存储实现
|
|
|
|
|
|
用于演示的内存存储
|
|
|
"""
|
|
|
|
|
|
def __init__(self):
|
|
|
"""初始化内存存储"""
|
|
|
self._data: Dict[str, Dict[str, Any]] = {}
|
|
|
self._lock = threading.RLock()
|
|
|
|
|
|
def get(self, entity_id: str) -> Optional[Dict[str, Any]]:
|
|
|
"""获取实体"""
|
|
|
with self._lock:
|
|
|
return self._data.get(entity_id)
|
|
|
|
|
|
def put(self, entity_id: str, data: Dict[str, Any]) -> None:
|
|
|
"""存储实体"""
|
|
|
with self._lock:
|
|
|
self._data[entity_id] = data
|
|
|
|
|
|
def delete(self, entity_id: str) -> bool:
|
|
|
"""删除实体"""
|
|
|
with self._lock:
|
|
|
if entity_id in self._data:
|
|
|
del self._data[entity_id]
|
|
|
return True
|
|
|
return False
|
|
|
|
|
|
def get_all(self) -> Dict[str, Dict[str, Any]]:
|
|
|
"""获取所有实体"""
|
|
|
with self._lock:
|
|
|
return self._data.copy()
|
|
|
|
|
|
|
|
|
class Node:
|
|
|
"""
|
|
|
系统节点类
|
|
|
|
|
|
表示分布式系统中的一个节点
|
|
|
"""
|
|
|
|
|
|
def __init__(self, node_id: str, storage: StorageProvider):
|
|
|
"""
|
|
|
初始化节点
|
|
|
|
|
|
Args:
|
|
|
node_id: 节点ID
|
|
|
storage: 存储提供者
|
|
|
"""
|
|
|
self.node_id = node_id
|
|
|
self.storage = storage
|
|
|
self.message_queue = MessageQueue(node_id)
|
|
|
self.received_messages: Dict[str, ReplicationMessage] = {}
|
|
|
self.sync_status: Dict[str, SyncStatus] = {}
|
|
|
self._lock = threading.RLock()
|
|
|
|
|
|
# 启动消息处理器线程
|
|
|
self._stop_event = threading.Event()
|
|
|
self._processor_thread = threading.Thread(target=self._process_messages)
|
|
|
self._processor_thread.daemon = True
|
|
|
self._processor_thread.start()
|
|
|
|
|
|
def create_entity(self, entity_id: str, data: Dict[str, Any]) -> ReplicationMessage:
|
|
|
"""
|
|
|
创建实体
|
|
|
|
|
|
Args:
|
|
|
entity_id: 实体ID
|
|
|
data: 实体数据
|
|
|
|
|
|
Returns:
|
|
|
复制消息
|
|
|
"""
|
|
|
# 添加元数据
|
|
|
data = {
|
|
|
**data,
|
|
|
"_version": 1,
|
|
|
"_created_at": time.time(),
|
|
|
"_updated_at": time.time(),
|
|
|
"_created_by": self.node_id
|
|
|
}
|
|
|
|
|
|
# 存储数据
|
|
|
self.storage.put(entity_id, data)
|
|
|
|
|
|
# 创建复制消息
|
|
|
message = ReplicationMessage("create", entity_id, data, self.node_id)
|
|
|
|
|
|
return message
|
|
|
|
|
|
def update_entity(self, entity_id: str, data: Dict[str, Any]) -> Optional[ReplicationMessage]:
|
|
|
"""
|
|
|
更新实体
|
|
|
|
|
|
Args:
|
|
|
entity_id: 实体ID
|
|
|
data: 更新的数据
|
|
|
|
|
|
Returns:
|
|
|
复制消息，如果实体不存在则返回None
|
|
|
"""
|
|
|
# 获取当前数据
|
|
|
current_data = self.storage.get(entity_id)
|
|
|
if not current_data:
|
|
|
return None
|
|
|
|
|
|
# 合并数据并更新元数据
|
|
|
updated_data = {
|
|
|
**current_data,
|
|
|
**{k: v for k, v in data.items() if not k.startswith('_')},
|
|
|
"_version": current_data["_version"] + 1,
|
|
|
"_updated_at": time.time()
|
|
|
}
|
|
|
|
|
|
# 存储更新后的数据
|
|
|
self.storage.put(entity_id, updated_data)
|
|
|
|
|
|
# 创建复制消息
|
|
|
message = ReplicationMessage("update", entity_id, updated_data, self.node_id)
|
|
|
|
|
|
return message
|
|
|
|
|
|
def delete_entity(self, entity_id: str) -> Optional[ReplicationMessage]:
|
|
|
"""
|
|
|
删除实体
|
|
|
|
|
|
Args:
|
|
|
entity_id: 实体ID
|
|
|
|
|
|
Returns:
|
|
|
复制消息，如果实体不存在则返回None
|
|
|
"""
|
|
|
# 获取当前数据
|
|
|
current_data = self.storage.get(entity_id)
|
|
|
if not current_data:
|
|
|
return None
|
|
|
|
|
|
# 删除数据
|
|
|
self.storage.delete(entity_id)
|
|
|
|
|
|
# 创建复制消息
|
|
|
message = ReplicationMessage("delete", entity_id, current_data, self.node_id)
|
|
|
|
|
|
return message
|
|
|
|
|
|
def receive_message(self, message: ReplicationMessage) -> bool:
|
|
|
"""
|
|
|
接收复制消息
|
|
|
|
|
|
Args:
|
|
|
message: 复制消息
|
|
|
|
|
|
Returns:
|
|
|
是否成功处理消息
|
|
|
"""
|
|
|
with self._lock:
|
|
|
# 检查是否已经处理过该消息
|
|
|
if message.message_id in self.received_messages:
|
|
|
return True
|
|
|
|
|
|
# 检查版本号，确保处理最新的消息
|
|
|
current_data = self.storage.get(message.entity_id)
|
|
|
if current_data and "_version" in current_data:
|
|
|
message_version = message.data.get("_version", 1)
|
|
|
if message_version <= current_data["_version"]:
|
|
|
print(f"Node {self.node_id}: Ignoring outdated message {message.message_id}")
|
|
|
return True
|
|
|
|
|
|
# 将消息加入队列
|
|
|
self.message_queue.enqueue(message)
|
|
|
return True
|
|
|
|
|
|
def _process_messages(self) -> None:
|
|
|
"""
|
|
|
处理消息队列中的消息
|
|
|
"""
|
|
|
while not self._stop_event.is_set():
|
|
|
try:
|
|
|
message = self.message_queue.dequeue(timeout=1.0)
|
|
|
if message:
|
|
|
self._apply_message(message)
|
|
|
except Exception as e:
|
|
|
print(f"Node {self.node_id}: Error processing message: {str(e)}")
|
|
|
|
|
|
def _apply_message(self, message: ReplicationMessage) -> None:
|
|
|
"""
|
|
|
应用消息到本地存储
|
|
|
|
|
|
Args:
|
|
|
message: 复制消息
|
|
|
"""
|
|
|
try:
|
|
|
with self._lock:
|
|
|
print(f"Node {self.node_id}: Applying message {message.message_id} ({message.operation})")
|
|
|
|
|
|
if message.operation == "create" or message.operation == "update":
|
|
|
# 存储数据
|
|
|
self.storage.put(message.entity_id, message.data)
|
|
|
elif message.operation == "delete":
|
|
|
# 删除数据
|
|
|
self.storage.delete(message.entity_id)
|
|
|
|
|
|
# 记录已接收的消息
|
|
|
self.received_messages[message.message_id] = message
|
|
|
|
|
|
# 更新同步状态
|
|
|
self.sync_status[message.entity_id] = SyncStatus.COMPLETED
|
|
|
|
|
|
return True
|
|
|
|
|
|
except Exception as e:
|
|
|
print(f"Node {self.node_id}: Failed to apply message {message.message_id}: {str(e)}")
|
|
|
with self._lock:
|
|
|
self.sync_status[message.entity_id] = SyncStatus.FAILED
|
|
|
return False
|
|
|
|
|
|
def get_entity(self, entity_id: str) -> Optional[Dict[str, Any]]:
|
|
|
"""
|
|
|
获取实体
|
|
|
|
|
|
Args:
|
|
|
entity_id: 实体ID
|
|
|
|
|
|
Returns:
|
|
|
实体数据，如果不存在则返回None
|
|
|
"""
|
|
|
return self.storage.get(entity_id)
|
|
|
|
|
|
def get_all_entities(self) -> Dict[str, Dict[str, Any]]:
|
|
|
"""
|
|
|
获取所有实体
|
|
|
|
|
|
Returns:
|
|
|
所有实体数据
|
|
|
"""
|
|
|
return self.storage.get_all()
|
|
|
|
|
|
def shutdown(self) -> None:
|
|
|
"""关闭节点"""
|
|
|
self._stop_event.set()
|
|
|
if self._processor_thread.is_alive():
|
|
|
self._processor_thread.join(timeout=2.0)
|
|
|
|
|
|
|
|
|
class ReplicationManager:
|
|
|
"""
|
|
|
复制管理器
|
|
|
|
|
|
负责管理节点间的数据复制
|
|
|
"""
|
|
|
|
|
|
def __init__(self):
|
|
|
"""初始化复制管理器"""
|
|
|
self.nodes: Dict[str, Node] = {}
|
|
|
self._lock = threading.RLock()
|
|
|
|
|
|
# 启动复制监控线程
|
|
|
self._stop_event = threading.Event()
|
|
|
self._monitor_thread = threading.Thread(target=self._monitor_replication)
|
|
|
self._monitor_thread.daemon = True
|
|
|
self._monitor_thread.start()
|
|
|
|
|
|
def add_node(self, node: Node) -> None:
|
|
|
"""
|
|
|
添加节点
|
|
|
|
|
|
Args:
|
|
|
node: 要添加的节点
|
|
|
"""
|
|
|
with self._lock:
|
|
|
self.nodes[node.node_id] = node
|
|
|
print(f"Added node {node.node_id}")
|
|
|
|
|
|
def remove_node(self, node_id: str) -> bool:
|
|
|
"""
|
|
|
移除节点
|
|
|
|
|
|
Args:
|
|
|
node_id: 要移除的节点ID
|
|
|
|
|
|
Returns:
|
|
|
是否移除成功
|
|
|
"""
|
|
|
with self._lock:
|
|
|
if node_id in self.nodes:
|
|
|
node = self.nodes.pop(node_id)
|
|
|
node.shutdown()
|
|
|
print(f"Removed node {node_id}")
|
|
|
return True
|
|
|
return False
|
|
|
|
|
|
def replicate_message(self, message: ReplicationMessage, exclude_node: Optional[str] = None) -> None:
|
|
|
"""
|
|
|
复制消息到所有其他节点
|
|
|
|
|
|
Args:
|
|
|
message: 要复制的消息
|
|
|
exclude_node: 要排除的节点ID
|
|
|
"""
|
|
|
with self._lock:
|
|
|
for node_id, node in self.nodes.items():
|
|
|
if node_id != exclude_node and node_id != message.source_node:
|
|
|
message.destination_nodes.add(node_id)
|
|
|
# 模拟网络延迟和故障
|
|
|
if random.random() < 0.9: # 90%的概率成功发送
|
|
|
node.receive_message(message)
|
|
|
else:
|
|
|
print(f"Failed to send message {message.message_id} to node {node_id}")
|
|
|
message.mark_failed(node_id)
|
|
|
|
|
|
def _monitor_replication(self) -> None:
|
|
|
"""
|
|
|
监控复制状态并处理失败的复制
|
|
|
"""
|
|
|
while not self._stop_event.is_set():
|
|
|
try:
|
|
|
# 这里可以实现失败消息的重试逻辑
|
|
|
# 为了简化，我们省略具体实现
|
|
|
time.sleep(5.0)
|
|
|
except Exception as e:
|
|
|
print(f"Replication manager error: {str(e)}")
|
|
|
|
|
|
def shutdown(self) -> None:
|
|
|
"""关闭复制管理器"""
|
|
|
self._stop_event.set()
|
|
|
if self._monitor_thread.is_alive():
|
|
|
self._monitor_thread.join(timeout=2.0)
|
|
|
|
|
|
# 关闭所有节点
|
|
|
for node in list(self.nodes.values()):
|
|
|
node.shutdown()
|
|
|
self.nodes.clear()
|
|
|
|
|
|
|
|
|
class DataSyncSystem:
|
|
|
"""
|
|
|
数据同步系统
|
|
|
|
|
|
整合节点和复制管理器，提供最终一致性的数据同步
|
|
|
"""
|
|
|
|
|
|
def __init__(self):
|
|
|
"""初始化数据同步系统"""
|
|
|
self.replication_manager = ReplicationManager()
|
|
|
|
|
|
def create_node(self, node_id: str, storage: Optional[StorageProvider] = None) -> Node:
|
|
|
"""
|
|
|
创建并添加节点
|
|
|
|
|
|
Args:
|
|
|
node_id: 节点ID
|
|
|
storage: 存储提供者，默认为内存存储
|
|
|
|
|
|
Returns:
|
|
|
创建的节点
|
|
|
"""
|
|
|
if storage is None:
|
|
|
storage = InMemoryStorage()
|
|
|
|
|
|
node = Node(node_id, storage)
|
|
|
self.replication_manager.add_node(node)
|
|
|
return node
|
|
|
|
|
|
def perform_write_operation(self, node_id: str, operation: str, entity_id: str, data: Dict[str, Any]) -> bool:
|
|
|
"""
|
|
|
在指定节点执行写操作
|
|
|
|
|
|
Args:
|
|
|
node_id: 节点ID
|
|
|
operation: 操作类型 (create, update, delete)
|
|
|
entity_id: 实体ID
|
|
|
data: 实体数据
|
|
|
|
|
|
Returns:
|
|
|
是否操作成功
|
|
|
"""
|
|
|
node = self.replication_manager.nodes.get(node_id)
|
|
|
if not node:
|
|
|
print(f"Node {node_id} not found")
|
|
|
return False
|
|
|
|
|
|
try:
|
|
|
# 在本地执行操作
|
|
|
if operation == "create":
|
|
|
message = node.create_entity(entity_id, data)
|
|
|
elif operation == "update":
|
|
|
message = node.update_entity(entity_id, data)
|
|
|
elif operation == "delete":
|
|
|
message = node.delete_entity(entity_id)
|
|
|
else:
|
|
|
print(f"Unknown operation: {operation}")
|
|
|
return False
|
|
|
|
|
|
if not message:
|
|
|
print(f"Operation failed: {operation} on {entity_id}")
|
|
|
return False
|
|
|
|
|
|
# 复制到其他节点
|
|
|
self.replication_manager.replicate_message(message)
|
|
|
|
|
|
return True
|
|
|
|
|
|
except Exception as e:
|
|
|
print(f"Error performing {operation}: {str(e)}")
|
|
|
return False
|
|
|
|
|
|
def get_entity_from_node(self, node_id: str, entity_id: str) -> Optional[Dict[str, Any]]:
|
|
|
"""
|
|
|
从指定节点获取实体
|
|
|
|
|
|
Args:
|
|
|
node_id: 节点ID
|
|
|
entity_id: 实体ID
|
|
|
|
|
|
Returns:
|
|
|
实体数据，如果不存在则返回None
|
|
|
"""
|
|
|
node = self.replication_manager.nodes.get(node_id)
|
|
|
if not node:
|
|
|
return None
|
|
|
|
|
|
return node.get_entity(entity_id)
|
|
|
|
|
|
def get_all_entities_from_node(self, node_id: str) -> Dict[str, Dict[str, Any]]:
|
|
|
"""
|
|
|
从指定节点获取所有实体
|
|
|
|
|
|
Args:
|
|
|
node_id: 节点ID
|
|
|
|
|
|
Returns:
|
|
|
所有实体数据
|
|
|
"""
|
|
|
node = self.replication_manager.nodes.get(node_id)
|
|
|
if not node:
|
|
|
return {}
|
|
|
|
|
|
return node.get_all_entities()
|
|
|
|
|
|
def shutdown(self) -> None:
|
|
|
"""关闭数据同步系统"""
|
|
|
self.replication_manager.shutdown()
|
|
|
|
|
|
|
|
|
# 示例使用
|
|
|
if __name__ == "__main__":
|
|
|
# 创建数据同步系统
|
|
|
sync_system = DataSyncSystem()
|
|
|
|
|
|
# 创建三个节点
|
|
|
node1 = sync_system.create_node("node-1")
|
|
|
node2 = sync_system.create_node("node-2")
|
|
|
node3 = sync_system.create_node("node-3")
|
|
|
|
|
|
print("\n=== 测试场景 1: 基本数据复制 ===")
|
|
|
|
|
|
# 在节点1创建实体
|
|
|
print("在节点1创建用户实体...")
|
|
|
sync_system.perform_write_operation(
|
|
|
"node-1",
|
|
|
"create",
|
|
|
"user-1",
|
|
|
{"username": "john_doe", "email": "john@example.com"}
|
|
|
)
|
|
|
|
|
|
# 等待复制完成
|
|
|
time.sleep(1)
|
|
|
|
|
|
# 检查各节点数据
|
|
|
print("\n各节点数据状态:")
|
|
|
for node_id in ["node-1", "node-2", "node-3"]:
|
|
|
user = sync_system.get_entity_from_node(node_id, "user-1")
|
|
|
status = "✓" if user else "✗"
|
|
|
print(f"节点 {node_id}: {status} {user}")
|
|
|
|
|
|
print("\n=== 测试场景 2: 更新数据 ===")
|
|
|
|
|
|
# 在节点2更新实体
|
|
|
print("在节点2更新用户邮箱...")
|
|
|
sync_system.perform_write_operation(
|
|
|
"node-2",
|
|
|
"update",
|
|
|
"user-1",
|
|
|
{"email": "john.doe@example.com"}
|
|
|
)
|
|
|
|
|
|
# 等待复制完成
|
|
|
time.sleep(1)
|
|
|
|
|
|
# 检查各节点更新后的数据
|
|
|
print("\n各节点更新后的数据状态:")
|
|
|
for node_id in ["node-1", "node-2", "node-3"]:
|
|
|
user = sync_system.get_entity_from_node(node_id, "user-1")
|
|
|
print(f"节点 {node_id}: 邮箱={user['email']}, 版本={user['_version']}")
|
|
|
|
|
|
print("\n=== 测试场景 3: 删除数据 ===")
|
|
|
|
|
|
# 在节点3删除实体
|
|
|
print("在节点3删除用户实体...")
|
|
|
sync_system.perform_write_operation(
|
|
|
"node-3",
|
|
|
"delete",
|
|
|
"user-1",
|
|
|
{}
|
|
|
)
|
|
|
|
|
|
# 等待复制完成
|
|
|
time.sleep(1)
|
|
|
|
|
|
# 检查各节点删除后的数据
|
|
|
print("\n各节点删除后的数据状态:")
|
|
|
for node_id in ["node-1", "node-2", "node-3"]:
|
|
|
user = sync_system.get_entity_from_node(node_id, "user-1")
|
|
|
status = "已删除" if user is None else "仍存在"
|
|
|
print(f"节点 {node_id}: {status}")
|
|
|
|
|
|
print("\n=== 测试场景 4: 多实体操作 ===")
|
|
|
|
|
|
# 创建多个实体
|
|
|
entities = [
|
|
|
("product-1", {"name": "Laptop", "price": 999.99}),
|
|
|
("product-2", {"name": "Smartphone", "price": 499.99}),
|
|
|
("product-3", {"name": "Tablet", "price": 299.99})
|
|
|
]
|
|
|
|
|
|
for entity_id, data in entities:
|
|
|
sync_system.perform_write_operation("node-1", "create", entity_id, data)
|
|
|
|
|
|
# 等待复制
|
|
|
time.sleep(1)
|
|
|
|
|
|
# 统计各节点的实体数量
|
|
|
print("\n各节点实体数量:")
|
|
|
for node_id in ["node-1", "node-2", "node-3"]:
|
|
|
entities = sync_system.get_all_entities_from_node(node_id)
|
|
|
print(f"节点 {node_id}: {len(entities)} 个实体")
|
|
|
|
|
|
# 关闭系统
|
|
|
sync_system.shutdown()
|
|
|
print("\n数据同步系统已关闭") |
