Java分布式数据备忘录模式等怎么备忘录

wen java案例 2026-07-04 213

本文目录导读：

备忘录模式基础概念
分布式环境下的挑战
分布式备忘录模式实现
最佳实践建议

我来详细解释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);
        }
    }
}

分布式备忘录模式的关键点：