01. 分布式系统理论基础

// 订单创建：先写入订单，异步扣减库存
@Transactional
public Order createOrder(OrderRequest request) {
    // 1. 创建订单（主库，强一致）
    Order order = orderRepository.save(new Order(request));
    
    // 2. 发送库存扣减消息（异步，最终一致）
    kafkaTemplate.send("inventory-deduct", 
        new InventoryEvent(order.getId(), request.getSku(), request.getQuantity()));
    
    // 3. 发送延迟消息，15分钟后检查订单状态
    kafkaTemplate.send("order-timeout-check", 
        new DelayedMessage(order.getId(), 15 * 60 * 1000));
    
    return order;
}

// 库存服务消费消息
@KafkaListener(topics = "inventory-deduct")
public void deductInventory(InventoryEvent event) {
    inventoryService.decrease(event.getSku(), event.getQuantity());
}

;; Jepsen 测试示例
(def db (db "etcd"))
(jepsen/run! (assoc tests/noop-test
  :name "etcd-linearizable"
  :db db
  :client (client nil)
  :checker (checker/linearizable)
  :nemesis (nemesis/partition-random-halves)))

def lamport_clock(events):
    """Lamport 标量时钟：偏序关系，无法识别并发事件"""
    clock = {}
    for process, event_type in events:
        if process not in clock:
            clock[process] = 0
        if event_type == 'send':
            clock[process] += 1
            # 发送消息携带当前时间戳
        elif event_type == 'receive':
            # 接收方: max(本地时钟, 消息时间戳) + 1
            clock[process] = max(clock[process], msg_timestamp) + 1
        else:  # local event
            clock[process] += 1
    return clock

class VectorClock:
    def __init__(self, num_processes, process_id):
        self.id = process_id
        self.clock = [0] * num_processes
    
    def increment(self):
        self.clock[self.id] += 1
    
    def update(self, other_clock):
        self.clock = [max(a, b) for a, b in zip(self.clock, other_clock)]
        self.increment()
    
    def compare(self, other):
        """返回: before, after, concurrent"""
        less = all(a <= b for a, b in zip(self.clock, other.clock))
        greater = all(a >= b for a, b in zip(self.clock, other.clock))
        if less and not greater:
            return "before"
        if greater and not less:
            return "after"
        if self.clock == other:
            return "equal"
        return "concurrent"

class DynamoVC:
    def __init__(self):
        self.versions = {}  # {node_id: counter}
    
    def increment(self, node_id):
        self.versions[node_id] = self.versions.get(node_id, 0) + 1
    
    def merge(self, other):
        for node, count in other.versions.items():
            self.versions[node] = max(self.versions.get(node, 0), count)
    
    def is_descendant(self, other):
        """self 是否是 other 的后继版本"""
        return all(self.versions.get(n, 0) >= c 
                   for n, c in other.versions.items())
    
    def has_conflict(self, other):
        return not (self.is_descendant(other) or other.is_descendant(self))