本文目录导读：
我来详细解释Java分布式系统中的备忘录模式及其实现方式。
备忘录模式基础概念
备忘录模式（Memento Pattern）用于保存对象的内部状态，以便在需要时恢复,包含三个角色：
- Originator（发起者）：需要保存状态的对象
- Memento（备忘录）：保存状态的对象
- Caretaker（管理者）：管理备忘录的存储和恢复
分布式环境下的挑战
在分布式系统中,传统备忘录模式面临挑战：
- 数据一致性
- 序列化/反序列化
- 网络传输延迟
- 状态同步
分布式备忘录模式实现
基础备忘录实现
// 发起者
public class OrderService {
private String orderId;
private String status;
private List<String> items;
private Map<String, Object> metadata;
public OrderMemento save() {
return new OrderMemento(
orderId,
status,
new ArrayList<>(items),
new HashMap<>(metadata)
);
}
public void restore(OrderMemento memento) {
this.orderId = memento.getOrderId();
this.status = memento.getStatus();
this.items = new ArrayList<>(memento.getItems());
this.metadata = new HashMap<>(memento.getMetadata());
}
}
// 备忘录 - 需要序列化
public class OrderMemento implements Serializable {
private static final long serialVersionUID = 1L;
private final String orderId;
private final String status;
private final List<String> items;
private final Map<String, Object> metadata;
private final long timestamp;
public OrderMemento(String orderId, String status,
List<String> items, Map<String, Object> metadata) {
this.orderId = orderId;
this.status = status;
this.items = items;
this.metadata = metadata;
this.timestamp = System.currentTimeMillis();
}
// getters...
}
// 管理者
public class MementoManager {
private final Map<String, Stack<OrderMemento>> mementoStore = new ConcurrentHashMap<>();
public void save(String orderId, OrderMemento memento) {
mementoStore.computeIfAbsent(orderId, k -> new Stack<>())
.push(memento);
}
public OrderMemento restore(String orderId) {
Stack<OrderMemento> stack = mementoStore.get(orderId);
return stack.isEmpty() ? null : stack.pop();
}
}
分布式缓存实现
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
@Service
public class DistributedMementoManager {
private final RedisTemplate<String, Object> redisTemplate;
private final String MEMENTO_PREFIX = "memento:";
public void saveMemento(String key, Serializable memento) {
String redisKey = MEMENTO_PREFIX + key;
// 使用List存储历史状态
redisTemplate.opsForList().leftPush(redisKey, memento);
// 限制历史记录数量
redisTemplate.opsForList().trim(redisKey, 0, MAX_MEMENTO_COUNT - 1);
}
public <T> T restoreMemento(String key) {
String redisKey = MEMENTO_PREFIX + key;
return (T) redisTemplate.opsForList().leftPop(redisKey);
}
public <T> List<T> getMementoHistory(String key) {
String redisKey = MEMENTO_PREFIX + key;
return (List<T>) redisTemplate.opsForList().range(redisKey, 0, -1);
}
}
分布式事务备忘录
// 分布式事务状态保存
public class DistributedTransactionMemento {
private final String transactionId;
private final String serviceId;
private final TransactionPhase phase;
private final Map<String, Serializable> serviceStates;
private final long timestamp;
// 序列化用于网络传输
public byte[] serialize() {
try {
ByteArrayOutputStream bos = new ByteArrayOutputStream();
ObjectOutputStream oos = new ObjectOutputStream(bos);
oos.writeObject(this);
return bos.toByteArray();
} catch (IOException e) {
throw new RuntimeException("Serialization failed", e);
}
}
public static DistributedTransactionMemento deserialize(byte[] data) {
try {
ByteArrayInputStream bis = new ByteArrayInputStream(data);
ObjectInputStream ois = new ObjectInputStream(bis);
return (DistributedTransactionMemento) ois.readObject();
} catch (Exception e) {
throw new RuntimeException("Deserialization failed", e);
}
}
}
// 分布式事务恢复
@Component
public class TransactionRecoveryService {
private final DistributedLockManager lockManager;
public boolean recoverTransaction(String transactionId) {
// 获取分布式锁
String lockKey = "recovery:" + transactionId;
if (lockManager.tryLock(lockKey, 30, TimeUnit.SECONDS)) {
try {
// 获取事务备忘录
DistributedTransactionMemento memento =
getMemento(transactionId);
if (memento == null) {
return false;
}
// 根据阶段恢复
switch (memento.getPhase()) {
case PREPARE:
return rollbackTransaction(transactionId);
case COMMIT:
return completeTransaction(transactionId);
case ROLLBACK:
return true; // 已回滚
}
return true;
} finally {
lockManager.unlock(lockKey);
}
}
return false;
}
}
基于事件溯源的备忘录
// 事件基类
public abstract class DomainEvent implements Serializable {
private final String eventId;
private final String aggregateId;
private final long timestamp;
// 构造函数和getters...
}
// 事件存储
@Component
public class EventStore {
private final MongoTemplate mongoTemplate;
public void saveEvent(DomainEvent event) {
mongoTemplate.save(event, getCollectionName(event));
}
public List<DomainEvent> getEvents(String aggregateId) {
Query query = Query.query(Criteria.where("aggregateId").is(aggregateId))
.with(Sort.by(Sort.Direction.ASC, "timestamp"));
return mongoTemplate.find(query, DomainEvent.class);
}
// 通过重放事件恢复状态
public <T> T replayEvents(String aggregateId, T initialState) {
List<DomainEvent> events = getEvents(aggregateId);
T currentState = initialState;
for (DomainEvent event : events) {
if (event instanceof OrderCreatedEvent) {
// 应用事件到状态
applyOrderCreated(currentState, (OrderCreatedEvent) event);
} else if (event instanceof OrderUpdatedEvent) {
applyOrderUpdated(currentState, (OrderUpdatedEvent) event);
}
}
return currentState;
}
}
高可用集群实现
@Component
public class ClusterMementoManager {
private final LoadBalancerClient loadBalancer;
private final DiscoveryClient discoveryClient;
// 跨节点状态同步
public void syncMementoToCluster(String key, Serializable memento) {
List<ServiceInstance> instances = discoveryClient.getInstances("memento-service");
// 并行发送到所有节点
CompletableFuture<?>[] futures = instances.stream()
.map(instance -> CompletableFuture.runAsync(() ->
sendMementoToNode(key, memento, instance)))
.toArray(CompletableFuture[]::new);
// 等待多数节点确认
CompletableFuture.allOf(futures).join();
}
// 从集群恢复 (Quorum读取)
public Serializable restoreFromCluster(String key) {
List<ServiceInstance> instances = discoveryClient.getInstances("memento-service");
Map<Serializable, Integer> versionCount = new ConcurrentHashMap<>();
// 从多个节点读取
instances.parallelStream()
.forEach(instance -> {
Serializable memento = getMementoFromNode(key, instance);
if (memento != null) {
versionCount.merge(memento, 1, Integer::sum);
}
});
// 返回出现次数最多的版本
return versionCount.entrySet().stream()
.max(Map.Entry.comparingByValue())
.map(Map.Entry::getKey)
.orElse(null);
}
}
最佳实践建议
性能优化
// 使用缓存减少数据库访问
@Component
public class OptimizedMementoManager {
private final LoadingCache<String, Stack<Memento>> cache;
public OptimizedMementoManager() {
this.cache = Caffeine.newBuilder()
.maximumSize(10000)
.expireAfterWrite(30, TimeUnit.MINUTES)
.build(key -> new Stack<>());
}
}
版本控制
// 版本化备忘录
public class VersionedMemento implements Serializable {
private final int version;
private final byte[] data;
private final String checksum;
public boolean validate() {
return computeChecksum(data).equals(checksum);
}
}
监控和告警
@Aspect
@Component
public class MementoMonitor {
@Around("@annotation(MementoOperation)")
public Object monitorOperation(ProceedingJoinPoint pjp) throws Throwable {
long start = System.currentTimeMillis();
try {
return pjp.proceed();
} finally {
long duration = System.currentTimeMillis() - start;
if (duration > THRESHOLD_MS) {
log.warn("Memento operation took too long: {}ms", duration);
}
metricsCollector.recordMementoOperation(duration);
}
}
}
分布式备忘录模式的关键点：
- 序列化兼容性 - 确保跨版本兼容
- 一致性保证 - 使用分布式锁或事务
- 性能优化 - 缓存和异步处理
- 容错处理 - 副本和恢复机制
- 监控追踪 - 记录操作和性能指标
根据具体业务场景，可以选择合适的实现方式,从简单的缓存方案到复杂的事件溯源架构。
