生产实践：etcd/Consul/ZooKeeper#

import io.etcd.jetcd.*;
import io.etcd.jetcd.kv.*;
import io.etcd.jetcd.lock.*;
import io.etcd.jetcd.lease.Lease;

import java.util.concurrent.TimeUnit;

/**
 * etcd Java 客户端使用示例
 * 使用 jetcd 库
 */
public class EtcdClientExample {

    private final Client client;
    private final KV kv;
    private final LockClient lockClient;

    public EtcdClientExample(String endpoints) {
        // 连接 etcd 集群
        this.client = Client.builder()
            .endpoints(endpoints.split(","))
            .build();
        this.kv = client.getKVClient();
        this.lockClient = client.getLockClient();
    }

    /**
     * 分布式锁实现
     */
    public LockResponse tryLock(String lockName, long ttlSeconds) throws Exception {
        // 创建租约
        Lease leaseClient = client.getLeaseClient();
        long leaseId = leaseClient.grant(ttlSeconds).get().getID();

        // 尝试获取锁
        LockOut lockOut = lockClient.lock(ByteSequence.from(lockName, StandardCharsets.UTF_8), leaseId).get();

        return new LockResponse(lockOut.getKey(), leaseId);
    }

    /**
     * 配置管理：Watch + 动态更新
     */
    public void watchConfig(String configPath, Consumer<String> onChange) {
        CountDownLatch latch = new CountDownLatch(1);

        Watch watchClient = client.getWatchClient();
        WatchListener listener = response -> {
            for (WatchEvent event : response.getEvents()) {
                switch (event.getEventType()) {
                    case PUT:
                        String newValue = event.getKeyValue().getValue().toString(StandardCharsets.UTF_8);
                        onChange.accept(newValue);
                        break;
                    case DELETE:
                        onChange.accept(null); // 配置被删除
                        break;
                }
            }
        };

        watchClient.watch(
            ByteSequence.from(configPath, StandardCharsets.UTF_8),
            listener
        );
    }

    /**
     * 分布式计数器（乐观锁）
     */
    public long incrementCounter(String counterPath) throws Exception {
        while (true) {
            // 获取当前值
            TxnResponse txnResponse = kv.txn()
                .If(new Version(ByteSequence.from(counterPath, StandardCharsets.UTF_8)).isNotExist()
                    .or(new Version(ByteSequence.from(counterPath, StandardCharsets.UTF_8)).gt(leaseId -> leaseId, TxnCompare.greaterThan, -1)))
                .Then(Op.put(ByteSequence.from(counterPath, StandardCharsets.UTF_8),
                           ByteSequence.from("1", StandardCharsets.UTF_8)))
                .Else(Op.put(ByteSequence.from(counterPath, StandardCharsets.UTF_8),
                            getValue(counterPath) + 1)))
                .commit();

            if (txnResponse.isSucceeded()) {
                return 1;
            } else {
                // 失败，重试
                long current = Long.parseLong(getValue(counterPath));
                kv.put(ByteSequence.from(counterPath, StandardCharsets.UTF_8),
                       ByteSequence.from(String.valueOf(current + 1), StandardCharsets.UTF_8)).get();
                return current + 1;
            }
        }
    }

    public void close() {
        client.close();
    }
}

// ZNode 类型
public enum ZNodeType {
    PERSISTENT,       // 持久节点：创建后一直存在，直到被删除
    PERSISTENT_SEQUENTIAL,  // 持久顺序节点：节点名会自动追加序号
    EPHEMERAL,        // 临时节点：创建它的会话断开后自动删除
    EPHEMERAL_SEQUENTIAL   // 临时顺序节点
}

import org.apache.zookeeper.*;

import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * ZooKeeper Java 客户端示例
 */
public class ZooKeeperExample implements Watcher {

    private ZooKeeper zk;
    private final CountDownLatch connectedSignal = new CountDownLatch(1);

    public void connect(String hosts) throws Exception {
        zk = new ZooKeeper(hosts, 5000, this);
        connectedSignal.await(); // 等待连接建立
    }

    @Override
    public void process(WatchedEvent event) {
        if (event.getState() == Event.KeeperState.SyncConnected) {
            connectedSignal.countDown();
        }
    }

    /**
     * 分布式锁实现
     */
    public boolean acquireLock(String lockPath, int timeoutMs) throws Exception {
        String lockNode = zk.create(
            lockPath + "/lock-",
            new byte[0],
            ZooDefs.Ids.OPEN_ACL_UNSAFE,
            CreateMode.EPHEMERAL_SEQUENTIAL
        );

        // 获取所有子节点
        List<String> children = zk.getChildren(lockPath, false);
        children.sort(String::compareTo);

        int myIndex = children.indexOf(lockNode.substring(lockNode.lastIndexOf('/') + 1));

        if (myIndex == 0) {
            // 我是最小的，获得锁
            return true;
        } else {
            // 监听前一个节点
            String previousNode = children.get(myIndex - 1);

            CountDownLatch latch = new CountDownLatch(1);

            // 一次性 Watch
            Stat stat = zk.exists(lockPath + "/" + previousNode, event -> {
                if (event.getType() == Event.EventType.NodeDeleted) {
                    latch.countDown();
                }
            });

            if (stat == null) {
                // 前一个节点已消失，直接获得锁
                return true;
            }

            // 等待前一个节点删除
            return latch.await(timeoutMs, TimeUnit.MILLISECONDS);
        }
    }

    /**
     * 分布式计数器
     */
    public long incrementCounter(String path) throws Exception {
        Stat stat = zk.exists(path, false);
        if (stat == null) {
            // 不存在，创建并设为 1
            zk.create(path, "1".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            return 1;
        } else {
            // 存在，乐观锁更新
            while (true) {
                int currentVersion = stat.getVersion();
                String value = new String(zk.getData(path, false, stat));
                long newValue = Long.parseLong(value) + 1;

                try {
                    zk.setData(path, String.valueOf(newValue).getBytes(), currentVersion);
                    return newValue;
                } catch (BadVersionException e) {
                    // 版本冲突，重试
                    stat = zk.exists(path, false);
                }
            }
        }
    }

    /**
     * Master 选举
     */
    public boolean tryBecomeMaster(String masterPath) throws Exception {
        try {
            zk.create(
                masterPath,
                ("master-" + zk.getSessionId()).getBytes(),
                ZooDefs.Ids.OPEN_ACL_UNSAFE,
                CreateMode.EPHEMERAL   // 临时节点
            );
            return true; // 创建成功，成为 Master
        } catch (NodeExistsException e) {
            return false; // 已被其他节点抢占
        }
    }

    public void watchMaster(String masterPath, Consumer<String> onMasterChange) throws Exception {
        CountDownLatch latch = new CountDownLatch(1);

        Stat stat = zk.exists(masterPath, event -> {
            if (event.getType() == Event.EventType.NodeDeleted) {
                onMasterChange.accept(null);
            }
        });

        if (stat != null) {
            byte[] data = zk.getData(masterPath, false, stat);
            onMasterChange.accept(new String(data));
        }
    }

    public void close() throws InterruptedException {
        zk.close();
    }
}

// Watch 使用模式
public class ConfigWatcher {

    private byte[] currentConfig;

    public void watchConfig(String configPath) throws Exception {
        // 第一次读取 + 注册 Watch
        readAndWatch(configPath);
    }

    private void readAndWatch(String path) throws Exception {
        // Watch 注册在读取操作中
        currentConfig = zk.getData(path, event -> {
            if (event.getType() == Event.EventType.NodeDataChanged) {
                // 数据变更，重新读取并注册新 Watch
                try {
                    readAndWatch(path);
                } catch (Exception e) {
                    // 处理异常
                }
            }
        }, null);
    }
}