BASE 理论：AP 系统的工程实践#

// 基本可用的降级示例
public ProductRecommendation getRecommendation(String userId) {
    try {
        // 尝试从主数据源获取
        return recommendationService.getFromDB(userId);
    } catch (DataAccessException e) {
        // 主数据源不可用，降级到缓存
        return recommendationService.getFromCache(userId);
    } catch (CacheException e) {
        // 缓存也不可用，降级到默认值
        return ProductRecommendation.DEFAULT;
    }
}

public class ReadRepair<K, V> {
    private final Map<K, Node> replicas;

    public V read(K key) {
        // 并发向所有副本发起读请求
        List<CompletableFuture<VersionedValue<V>>> futures = replicas.values().stream()
            .map(node -> node.readAsync(key))
            .collect(Collectors.toList());

        // 等待所有响应
        List<VersionedValue<V>> responses = futures.stream()
            .map(CompletableFuture::join)
            .collect(Collectors.toList());

        // 找到最新版本
        VersionedValue<V> latest = responses.stream()
            .max(Comparator.comparingLong(VersionedValue::getVersion))
            .orElseThrow(() -> new KeyNotFoundException(key));

        // 异步修复过期副本
        responses.stream()
            .filter(r -> r.getVersion() < latest.getVersion())
            .forEach(r -> r.getNode().writeAsync(key, latest.getValue(), latest.getVersion()));

        return latest.getValue();
    }
}

// Merkle Tree 原理：哈希树
// 叶子节点是单个键值对的哈希
// 父节点是子节点哈希的哈希
// 如果根哈希相同 → 两棵树完全相同
// 如果根哈希不同 → 只需同步差异分支

public class MerkleTree {
    // 层数计算：如果有 N 个键，Merkle Tree 的高度约为 log₂(N)
    // 10亿键 → 高度约 30 层
    // 对比根哈希只需要 1 次网络往返，而不是传输 10亿 条数据
}

public class HintedHandoff {
    private final Map<Node, List<HintedRecord>> pendingHints = new ConcurrentHashMap<>();

    // 节点B故障时，在节点A上记录一个提示
    public void storeHint(Node failedNode, Record record) {
        pendingHints.computeIfAbsent(failedNode, k -> new ArrayList<>())
            .add(new HintedRecord(record, System.currentTimeMillis()));
    }

    // 节点B恢复后，移交数据
    public void handoff(Node recoveredNode) {
        List<HintedRecord> hints = pendingHints.remove(recoveredNode);
        if (hints != null) {
            for (HintedRecord hint : hints) {
                recoveredNode.write(hint.getRecord());
            }
        }
    }
}

public class LWWConflictResolver<T> {
    public T resolve(List<VersionedValue<T>> versions) {
        return versions.stream()
            .max(Comparator.comparingLong(VersionedValue::getTimestamp))
            .map(VersionedValue::getValue)
            .orElse(null);
    }
}

// 向量时钟示例
// 版本1: {node_A: 1}         - A写入x=1
// 版本2: {node_A: 2, node_B: 1} - B也写入了x
// 通过比较向量时钟，可以判断版本之间是因果关系还是并发冲突

public class TombstoneExample {
    // 正常记录
    // key="order:123", value={...}, deleted=false

    // 删除操作 → 不物理删除，而是标记墓碑
    // key="order:123", value=null, deleted=true, tombstone_timestamp=...

    // 读取时跳过墓碑（逻辑删除）
    // GC 时物理删除墓碑
}

// 预扣减的伪代码
public class InventoryService {
    // 本地库存快照（软状态）
    private ConcurrentHashMap<String, AtomicInteger> localSnapshot = new ConcurrentHashMap<>();

    public boolean freeze(String itemId, int quantity) {
        // 本地操作，无网络往返
        AtomicInteger available = localSnapshot.computeIfAbsent(itemId, k -> new AtomicInteger(0));
        return available.addAndGet(-quantity) >= 0;
    }

    public void asyncConfirm(String itemId, int quantity) {
        // 写入队列，异步处理
        inventoryQueue.offer(new InventoryConfirmTask(itemId, quantity));
    }
}

public class SagaOrchestrator {
    private final List<Step> steps;
    private final Map<Step, CompensatingStep> compensations;

    public void execute() {
        List<Step> executedSteps = new ArrayList<>();
        try {
            for (Step step : steps) {
                step.execute();
                executedSteps.add(step);
            }
        } catch (Exception e) {
            // 回滚：逆序执行补偿操作
            for (int i = executedSteps.size() - 1; i >= 0; i--) {
                compensations.get(executedSteps.get(i)).compensate();
            }
        }
    }
}