/**
 * ZooKeeper分佈式鎖核心實現
 * 基於臨時順序節點和Watcher機制實現公平分佈式鎖
 */
public class ZkDistributedLock implements Watcher {
    private final ZooKeeper zookeeper;
    private final String lockBasePath;
    private final String lockName;
    private String currentLockPath;
    private CountDownLatch latch;
    
    private static final String LOCK_PREFIX = "/lock-";
    private static final int SESSION_TIMEOUT = 30000;
    
    public ZkDistributedLock(String zkAddress, String lockBasePath, String lockName) 
        throws IOException {
        this.zookeeper = new ZooKeeper(zkAddress, SESSION_TIMEOUT, this);
        this.lockBasePath = lockBasePath;
        this.lockName = lockName;
        ensureBasePath();
    }
    
    /**
     * 嘗試獲取分佈式鎖
     */
    public boolean tryLock(long timeout, TimeUnit unit) throws Exception {
        // 創建臨時順序節點
        currentLockPath = zookeeper.create(
            lockBasePath + LOCK_PREFIX, 
            new byte[0],
            ZooDefs.Ids.OPEN_ACL_UNSAFE,
            CreateMode.EPHEMERAL_SEQUENTIAL
        );
        
        // 獲取鎖，實現公平競爭
        return acquireLock(timeout, unit);
    }
    
    private boolean acquireLock(long timeout, TimeUnit unit) throws Exception {
        // 獲取所有鎖節點並排序
        List<String> allLocks = zookeeper.getChildren(lockBasePath, false);
        Collections.sort(allLocks);
        
        String currentLockName = currentLockPath.substring(lockBasePath.length() + 1);
        int currentIndex = allLocks.indexOf(currentLockName);
        
        // 當前節點是最小序號節點，獲得鎖
        if (currentIndex == 0) {
            return true;
        }
        
        // 監聽前一個節點
        String previousLockPath = lockBasePath + "/" + allLocks.get(currentIndex - 1);
        Stat stat = zookeeper.exists(previousLockPath, true);
        
        if (stat != null) {
            this.latch = new CountDownLatch(1);
            // 等待鎖釋放或超時
            return latch.await(timeout, unit);
        }
        
        // 前一個節點已不存在，重新嘗試獲取鎖
        return acquireLock(timeout, unit);
    }
    
    /**
     * 釋放分佈式鎖
     */
    public void unlock() throws Exception {
        if (currentLockPath != null) {
            zookeeper.delete(currentLockPath, -1);
            currentLockPath = null;
        }
    }
    
    @Override
    public void process(WatchedEvent event) {
        if (event.getType() == Event.EventType.NodeDeleted && latch != null) {
            latch.countDown(); // 前一個鎖節點被刪除，通知等待線程
        }
    }
    
    private void ensureBasePath() throws Exception {
        if (zookeeper.exists(lockBasePath, false) == null) {
            zookeeper.create(lockBasePath, new byte[0], 
                ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    }
}

/**
 * 基於Curator框架的分佈式鎖實現
 * Curator提供了更簡潔的API和更好的異常處理
 */
@Configuration
public class CuratorLockConfig {
    
    @Bean
    public CuratorFramework curatorFramework() {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework client = CuratorFrameworkFactory.newClient(
            "localhost:2181", retryPolicy);
        client.start();
        return client;
    }
    
    @Bean
    public InterProcessLock interProcessLock(CuratorFramework curatorFramework) {
        return new InterProcessMutex(curatorFramework, "/locks/distributed-lock");
    }
}

/**
 * 分佈式鎖服務
 */
@Service
@Slf4j
public class DistributedLockService {
    
    @Autowired
    private InterProcessLock interProcessLock;
    
    /**
     * 執行需要分佈式鎖保護的操作
     */
    public void executeWithLock(String businessKey, Runnable task) {
        boolean acquired = false;
        try {
            // 嘗試獲取鎖，最多等待5秒
            acquired = interProcessLock.acquire(5, TimeUnit.SECONDS);
            
            if (acquired) {
                log.info("成功獲取分佈式鎖，執行業務操作: {}", businessKey);
                task.run();
            } else {
                throw new LockAcquisitionException("獲取分佈式鎖超時");
            }
        } catch (Exception e) {
            throw new LockOperationException("分佈式鎖操作異常", e);
        } finally {
            if (acquired) {
                try {
                    interProcessLock.release();
                    log.info("釋放分佈式鎖: {}", businessKey);
                } catch (Exception e) {
                    log.warn("釋放分佈式鎖失敗", e);
                }
            }
        }
    }
    
    /**
     * 可重入鎖使用示例
     */
    public void reentrantLockExample() {
        try {
            // 第一次獲取鎖
            if (interProcessLock.acquire(10, TimeUnit.SECONDS)) {
                try {
                    // 第二次獲取同一把鎖（可重入）
                    if (interProcessLock.acquire(10, TimeUnit.SECONDS)) {
                        try {
                            // 業務邏輯
                            doBusiness();
                        } finally {
                            interProcessLock.release(); // 釋放第二次獲取的鎖
                        }
                    }
                } finally {
                    interProcessLock.release(); // 釋放第一次獲取的鎖
                }
            }
        } catch (Exception e) {
            throw new RuntimeException("可重入鎖操作失敗", e);
        }
    }
}

/**
 * ZooKeeper分佈式讀寫鎖實現
 * 支持多個讀鎖或一個寫鎖
 */
public class ZkReadWriteLock {
    private final InterProcessReadWriteLock readWriteLock;
    private InterProcessLock readLock;
    private InterProcessLock writeLock;
    
    public ZkReadWriteLock(CuratorFramework client, String lockPath) {
        this.readWriteLock = new InterProcessReadWriteLock(client, lockPath);
        this.readLock = readWriteLock.readLock();
        this.writeLock = readWriteLock.writeLock();
    }
    
    /**
     * 獲取讀鎖並執行操作
     */
    public <T> T executeWithReadLock(Callable<T> task, long timeout, TimeUnit unit) {
        boolean acquired = false;
        try {
            acquired = readLock.acquire(timeout, unit);
            if (acquired) {
                return task.call();
            }
            throw new LockTimeoutException("獲取讀鎖超時");
        } catch (Exception e) {
            throw new LockOperationException("讀鎖操作異常", e);
        } finally {
            if (acquired) {
                try {
                    readLock.release();
                } catch (Exception e) {
                    log.warn("釋放讀鎖失敗", e);
                }
            }
        }
    }
    
    /**
     * 獲取寫鎖並執行操作
     */
    public <T> T executeWithWriteLock(Callable<T> task, long timeout, TimeUnit unit) {
        boolean acquired = false;
        try {
            acquired = writeLock.acquire(timeout, unit);
            if (acquired) {
                return task.call();
            }
            throw new LockTimeoutException("獲取寫鎖超時");
        } catch (Exception e) {
            throw new LockOperationException("寫鎖操作異常", e);
        } finally {
            if (acquired) {
                try {
                    writeLock.release();
                } catch (Exception e) {
                    log.warn("釋放寫鎖失敗", e);
                }
            }
        }
    }
}

/**
 * 分佈式鎖監控服務
 * 實時監控鎖狀態，提供診斷信息
 */
@Service
@Slf4j
public class LockMonitorService {
    
    @Autowired
    private CuratorFramework curatorFramework;
    
    private final MeterRegistry meterRegistry;
    private final Timer lockAcquisitionTimer;
    private final Counter lockTimeoutCounter;
    
    public LockMonitorService(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
        this.lockAcquisitionTimer = Timer.builder("zookeeper.lock.acquisition.time")
            .description("Time taken to acquire distributed lock")
            .register(meterRegistry);
        
        this.lockTimeoutCounter = Counter.builder("zookeeper.lock.timeout.count")
            .description("Number of lock acquisition timeouts")
            .register(meterRegistry);
    }
    
    /**
     * 監控鎖競爭情況
     */
    @Scheduled(fixedRate = 30000)
    public void monitorLockContention() {
        try {
            List<String> locks = curatorFramework.getChildren().forPath("/locks");
            for (String lockPath : locks) {
                String fullPath = "/locks/" + lockPath;
                List<String> waiters = curatorFramework.getChildren().forPath(fullPath);
                
                Gauge.builder("zookeeper.lock.waiters.count", () -> waiters.size())
                    .tag("lock_path", fullPath)
                    .register(meterRegistry);
                
                if (waiters.size() > 10) {
                    log.warn("鎖競爭激烈: {} 有 {} 個等待者", fullPath, waiters.size());
                    alertService.sendAlert("鎖競爭激烈告警", fullPath);
                }
            }
        } catch (Exception e) {
            log.error("監控鎖競爭狀態失敗", e);
        }
    }
    
    /**
     * 記錄鎖獲取耗時
     */
    public void recordLockAcquisitionTime(long duration, TimeUnit unit) {
        lockAcquisitionTimer.record(duration, unit);
    }
    
    /**
     * 記錄鎖超時事件
     */
    public void recordLockTimeout() {
        lockTimeoutCounter.increment();
    }
    
    /**
     * 診斷鎖死鎖情況
     */
    public void diagnoseDeadlocks() {
        try {
            List<String> allLocks = curatorFramework.getChildren().forPath("/locks");
            for (String lockName : allLocks) {
                checkLockHealth("/locks/" + lockName);
            }
        } catch (Exception e) {
            log.error("診斷死鎖失敗", e);
        }
    }
    
    private void checkLockHealth(String lockPath) throws Exception {
        List<String> nodes = curatorFramework.getChildren().forPath(lockPath);
        if (nodes.size() > 1) {
            // 檢查是否有長時間持有的鎖
            Collections.sort(nodes);
            String firstNode = nodes.get(0);
            Stat stat = curatorFramework.checkExists().forPath(lockPath + "/" + firstNode);
            
            if (stat != null && System.currentTimeMillis() - stat.getCtime() > 300000) {
                log.warn("檢測到可能死鎖: {}", lockPath);
                alertService.sendAlert("死鎖檢測告警", lockPath);
            }
        }
    }
}

# ZooKeeper集羣配置
zookeeper:
  cluster:
    nodes:
      - server1:2181
      - server2:2181
      - server3:2181
  session:
    timeout: 30000
    connection: 
      timeout: 15000
  retry:
    baseSleepTime: 1000
    maxRetries: 3
    maxSleepTime: 10000

# 分佈式鎖配置
distributed:
  lock:
    basePath: /distributed-locks
    timeout:
      acquisition: 5000
      operation: 30000
    retry:
      policy: exponential
      maxAttempts: 3
    monitoring:
      enabled: true
      interval: 30000

/**
 * 分佈式鎖異常處理策略
 * 提供統一的異常處理和重試機制
 */
@Component
@Slf4j
public class LockExceptionHandler {
    
    private final RetryTemplate retryTemplate;
    
    public LockExceptionHandler() {
        this.retryTemplate = new RetryTemplate();
        
        ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
        backOffPolicy.setInitialInterval(1000);
        backOffPolicy.setMultiplier(2.0);
        backOffPolicy.setMaxInterval(10000);
        
        SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
        retryPolicy.setMaxAttempts(3);
        
        retryTemplate.setBackOffPolicy(backOffPolicy);
        retryTemplate.setRetryPolicy(retryPolicy);
        
        // 配置重試監聽器
        retryTemplate.registerListener(new RetryListener() {
            @Override
            public <T, E extends Throwable> void onError(RetryContext context, 
                RetryCallback<T, E> callback, Throwable throwable) {
                log.warn("分佈式鎖操作重試: 第{}次嘗試", context.getRetryCount(), throwable);
            }
        });
    }
    
    /**
     * 帶重試的鎖操作執行
     */
    public <T> T executeWithRetry(LockOperationCallback<T> callback) {
        return retryTemplate.execute(context -> {
            try {
                return callback.doInLock();
            } catch (KeeperException e) {
                if (e.code() == KeeperException.Code.CONNECTIONLOSS) {
                    throw new TransientLockException("ZooKeeper連接丟失", e);
                } else if (e.code() == KeeperException.Code.SESSIONEXPIRED) {
                    throw new TransientLockException("ZooKeeper會話過期", e);
                }
                throw new PermanentLockException("永久性鎖操作失敗", e);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new LockInterruptedException("鎖操作被中斷", e);
            }
        });
    }
    
    /**
     * 處理會話過期異常
     */
    public void handleSessionExpired() {
        log.error("ZooKeeper會話過期，需要重新建立連接");
        // 重新初始化ZooKeeper客户端
        reinitializeZookeeperClient();
        // 清理殘留的鎖狀態
        cleanupStaleLocks();
    }
    
    /**
     * 處理連接丟失異常
     */
    public void handleConnectionLoss() {
        log.warn("ZooKeeper連接丟失，嘗試重連");
        // 實現重連邏輯
        attemptReconnect();
    }
    
    public interface LockOperationCallback<T> {
        T doInLock() throws Exception;
    }
}