ZooKeeper配置管理

1. 配置文件结构 (application.yml)

zookeeper:
  # ZooKeeper服务器地址列表，多个地址用逗号分隔
  connect-string: localhost:2181
  # 会话超时时间（毫秒）
  session-timeout: 15000
  # 连接超时时间（毫秒）
  connection-timeout: 15000
  # 重试策略配置
  retry:
    max-attempts: 5
    base-sleep-time: 10000  # 毫秒
  # 根节点路径
  root-path: /myapp/config
  # 认证信息（格式：username:password）
  auth: admin:password123

# 应用配置
application:
  name: my-service

2. ZooKeeperConfigLoader类

负责加载和管理应用配置，替代原ConfigProperties类：

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.yaml.snakeyaml.Yaml;

import java.io.InputStream;
import java.util.Map;

/**
 * ZooKeeper配置加载器，负责从application.yml加载配置信息
 * 使用单例模式确保配置全局一致性
 */
public class ZooKeeperConfigLoader {
    private static final Logger logger = LoggerFactory.getLogger(ZooKeeperConfigLoader.class);
    private static final String CONFIG_FILE = "application.yml";
    
    // 配置数据缓存
    private final Map<String, Object> configData;
    private static final ZooKeeperConfigLoader INSTANCE = new ZooKeeperConfigLoader();

    private ZooKeeperConfigLoader() {
        try (InputStream inputStream = getClass().getClassLoader().getResourceAsStream(CONFIG_FILE)) {
            if (inputStream == null) {
                throw new IllegalArgumentException("配置文件 " + CONFIG_FILE + " 不存在");
            }
            Yaml yaml = new Yaml();
            configData = yaml.load(inputStream);
            logger.info("配置文件加载成功");
        } catch (Exception e) {
            logger.error("加载配置文件失败", e);
            throw new RuntimeException("初始化配置加载器失败", e);
        }
    }

    public static ZooKeeperConfigLoader getInstance() {
        return INSTANCE;
    }

    /**
     * 获取ZooKeeper连接字符串
     */
    public String getZkConnectString() {
        return getNestedValue("zookeeper.connect-string", String.class);
    }

    /**
     * 获取会话超时时间
     */
    public int getSessionTimeout() {
        return getNestedValue("zookeeper.session-timeout", Integer.class);
    }

    /**
     * 获取连接超时时间
     */
    public int getConnectionTimeout() {
        return getNestedValue("zookeeper.connection-timeout", Integer.class);
    }

    /**
     * 获取最大重试次数
     */
    public int getMaxRetries() {
        return getNestedValue("zookeeper.retry.max-attempts", Integer.class);
    }

    /**
     * 获取基础休眠时间
     */
    public int getBaseSleepTime() {
        return getNestedValue("zookeeper.retry.base-sleep-time", Integer.class);
    }

    /**
     * 获取ZooKeeper根路径
     */
    public String getRootPath() {
        return getNestedValue("zookeeper.root-path", String.class);
    }

    /**
     * 获取认证信息
     */
    public String getAuth() {
        return getNestedValue("zookeeper.auth", String.class);
    }

    /**
     * 获取应用名称
     */
    public String getApplicationName() {
        return getNestedValue("application.name", String.class);
    }

    /**
     * 从嵌套Map中获取值
     */
    @SuppressWarnings("unchecked")
    private <T> T getNestedValue(String path, Class<T> type) {
        String[] keys = path.split("\\.");
        Map<String, Object> currentMap = configData;
        
        for (int i = 0; i < keys.length - 1; i++) {
            Object value = currentMap.get(keys[i]);
            if (value == null || !(value instanceof Map)) {
                return null;
            }
            currentMap = (Map<String, Object>) value;
        }
        
        return type.cast(currentMap.get(keys[keys.length - 1]));
    }
}

3. ZooKeeperConfigManager类

负责配置的管理和缓存，替代原ConfigOne类：

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkState;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;

/**
 * ZooKeeper配置管理器，负责缓存配置、处理配置变更事件
 * 使用线程安全的数据结构确保多线程环境下的安全
 */
public class ZooKeeperConfigManager {
    private static final Logger logger = LoggerFactory.getLogger(ZooKeeperConfigManager.class);
    
    // 使用ConcurrentHashMap确保线程安全
    private static final Map<String, String> CONFIG_CACHE = new ConcurrentHashMap<>();
    private static final List<ConfigChangeListener> CONFIG_LISTENERS = new CopyOnWriteArrayList<>();
    private static final List<PathChangeListener> PATH_LISTENERS = new CopyOnWriteArrayList<>();
    
    private static final ZooKeeperConfigManager INSTANCE = new ZooKeeperConfigManager();
    
    private ZooKeeperConfigManager() {
        // 私有构造函数
    }
    
    public static ZooKeeperConfigManager getInstance() {
        return INSTANCE;
    }
    
    /**
     * 加载配置到缓存
     */
    public void loadConfig(String path, String key, String value) {
        logger.info("加载配置: {} = {}", key, value);
        CONFIG_CACHE.put(key, value);
    }
    
    /**
     * 获取配置值
     */
    public String getConfig(String key) {
        return CONFIG_CACHE.get(key);
    }
    
    /**
     * 获取配置值，支持默认值
     */
    public String getConfig(String key, String defaultValue) {
        return CONFIG_CACHE.getOrDefault(key, defaultValue);
    }
    
    /**
     * 添加配置变更监听器
     */
    public void addConfigListener(ConfigChangeListener listener) {
        CONFIG_LISTENERS.add(listener);
    }
    
    /**
     * 移除配置变更监听器
     */
    public void removeConfigListener(ConfigChangeListener listener) {
        CONFIG_LISTENERS.remove(listener);
    }
    
    /**
     * 添加路径变更监听器
     */
    public void addPathListener(PathChangeListener listener) {
        PATH_LISTENERS.add(listener);
    }
    
    /**
     * 移除路径变更监听器
     */
    public void removePathListener(PathChangeListener listener) {
        PATH_LISTENERS.remove(listener);
    }
    
    /**
     * 处理配置插入事件
     */
    public void handleConfigInsert(String path, String key, String value) {
        CONFIG_CACHE.put(key, value);
        logger.info("配置插入: {} = {}", key, value);
        
        for (ConfigChangeListener listener : CONFIG_LISTENERS) {
            try {
                listener.onConfigInserted(key, value);
            } catch (Exception e) {
                logger.error("处理配置插入事件失败", e);
            }
        }
    }
    
    /**
     * 处理配置更新事件
     */
    public void handleConfigUpdate(String path, String key, String newValue) {
        String oldValue = CONFIG_CACHE.put(key, newValue);
        logger.info("配置更新: {} = {} (旧值: {})", key, newValue, oldValue);
        
        for (ConfigChangeListener listener : CONFIG_LISTENERS) {
            try {
                listener.onConfigUpdated(key, oldValue, newValue);
            } catch (Exception e) {
                logger.error("处理配置更新事件失败", e);
            }
        }
    }
    
    /**
     * 处理配置删除事件
     */
    public void handleConfigDelete(String path, String key) {
        String oldValue = CONFIG_CACHE.remove(key);
        if (oldValue != null) {
            logger.info("配置删除: {}", key);
            
            for (ConfigChangeListener listener : CONFIG_LISTENERS) {
                try {
                    listener.onConfigDeleted(key, oldValue);
                } catch (Exception e) {
                    logger.error("处理配置删除事件失败", e);
                }
            }
        }
    }
    
    /**
     * 处理路径添加事件
     */
    public void handlePathAdded(String path) {
        logger.info("路径添加: {}", path);
        
        for (PathChangeListener listener : PATH_LISTENERS) {
            try {
                listener.onPathAdded(path);
            } catch (Exception e) {
                logger.error("处理路径添加事件失败", e);
            }
        }
    }
    
    /**
     * 处理路径删除事件
     */
    public void handlePathDeleted(String path) {
        logger.info("路径删除: {}", path);
        
        for (PathChangeListener listener : PATH_LISTENERS) {
            try {
                listener.onPathDeleted(path);
            } catch (Exception e) {
                logger.error("处理路径删除事件失败", e);
            }
        }
    }
    
    /**
     * 配置变更监听器接口
     */
    public interface ConfigChangeListener {
        void onConfigInserted(String key, String value);
        void onConfigUpdated(String key, String oldValue, String newValue);
        void onConfigDeleted(String key, String oldValue);
    }
    
    /**
     * 路径变更监听器接口
     */
    public interface PathChangeListener {
        void onPathAdded(String path);
        void onPathDeleted(String path);
    }
}

4. ZookeeperClient类

负责与ZooKeeper服务的通信，实现节点操作和监听：

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * ZooKeeper客户端，负责与ZooKeeper服务器通信
 * 使用单例模式确保全局只有一个连接实例
 */
public class ZookeeperClient {
    private static final Logger logger = LoggerFactory.getLogger(ZookeeperClient.class);
    private static final ExecutorService EVENT_EXECUTOR = Executors.newFixedThreadPool(10);
    
    private static volatile ZookeeperClient INSTANCE;
    private CuratorFramework client;
    private final ZooKeeperConfigLoader configLoader = ZooKeeperConfigLoader.getInstance();
    
    private ZookeeperClient() {
        init();
    }
    
    public static ZookeeperClient getInstance() {
        if (INSTANCE == null) {
            synchronized (ZookeeperClient.class) {
                if (INSTANCE == null) {
                    INSTANCE = new ZookeeperClient();
                }
            }
        }
        return INSTANCE;
    }
    
    /**
     * 初始化ZooKeeper客户端
     */
    private void init() {
        try {
            // 创建重试策略
            ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(
                configLoader.getBaseSleepTime(),
                configLoader.getMaxRetries()
            );
            
            // 创建客户端
            client = CuratorFrameworkFactory.builder()
                .connectString(configLoader.getZkConnectString())
                .sessionTimeoutMs(configLoader.getSessionTimeout())
                .connectionTimeoutMs(configLoader.getConnectionTimeout())
                .retryPolicy(retryPolicy)
                .build();
            
            // 添加认证信息
            String auth = configLoader.getAuth();
            if (auth != null && !auth.isEmpty()) {
                client.getZookeeperClient().getZooKeeper()
                    .addAuthInfo("digest", auth.getBytes(StandardCharsets.UTF_8));
            }
            
            // 启动客户端
            client.start();
            logger.info("ZooKeeper客户端启动成功");
            
            // 注册连接状态监听器
            client.getConnectionStateListenable().addListener((client, newState) -> {
                logger.info("ZooKeeper连接状态变更: {}", newState);
                if (newState.isConnected()) {
                    // 连接恢复，重新加载配置
                    reloadAllConfigs();
                }
            });
            
            // 注册根路径监听器
            registerRootPathListener();
            
        } catch (Exception e) {
            logger.error("初始化ZooKeeper客户端失败", e);
            throw new RuntimeException("初始化ZooKeeper客户端失败", e);
        }
    }
    
    /**
     * 重新加载所有配置
     */
    private void reloadAllConfigs() {
        try {
            ZooKeeperConfigManager configManager = ZooKeeperConfigManager.getInstance();
            String rootPath = configLoader.getRootPath();
            
            // 清空现有配置
            configManager.getAllConfigs().keySet().forEach(key -> 
                configManager.handleConfigDelete("", key)
            );
            
            // 递归加载配置
            loadConfigsRecursively(rootPath);
        } catch (Exception e) {
            logger.error("重新加载配置失败", e);
        }
    }
    
    /**
     * 递归加载配置
     */
    private void loadConfigsRecursively(String path) throws Exception {
        if (client.checkExists().forPath(path) == null) {
            return;
        }
        
        ZooKeeperConfigManager configManager = ZooKeeperConfigManager.getInstance();
        String rootPath = configLoader.getRootPath();
        
        // 获取节点数据
        byte[] data = client.getData().forPath(path);
        if (data != null && data.length > 0) {
            String key = path.substring(rootPath.length() + 1).replace("/", ".");
            String value = new String(data, StandardCharsets.UTF_8);
            configManager.loadConfig(path, key, value);
        }
        
        // 递归处理子节点
        for (String child : client.getChildren().forPath(path)) {
            loadConfigsRecursively(path + "/" + child);
        }
    }
    
    /**
     * 注册根路径监听器
     */
    private void registerRootPathListener() {
        try {
            String rootPath = configLoader.getRootPath();
            TreeCache treeCache = new TreeCache(client, rootPath);
            
            treeCache.getListenable().addListener((client, event) -> {
                TreeCacheEvent.Type eventType = event.getType();
                ChildData data = event.getData();
                
                if (data == null) {
                    return;
                }
                
                String path = data.getPath();
                String key = path.substring(configLoader.getRootPath().length() + 1).replace("/", ".");
                String value = data.getData() != null ? 
                    new String(data.getData(), StandardCharsets.UTF_8) : "";
                
                ZooKeeperConfigManager configManager = ZooKeeperConfigManager.getInstance();
                
                switch (eventType) {
                    case NODE_ADDED:
                        configManager.handleConfigInsert(path, key, value);
                        break;
                    case NODE_UPDATED:
                        configManager.handleConfigUpdate(path, key, value);
                        break;
                    case NODE_REMOVED:
                        configManager.handleConfigDelete(path, key);
                        break;
                    case CONNECTION_SUSPENDED:
                    case CONNECTION_RECONNECTED:
                    case CONNECTION_LOST:
                        logger.info("连接状态变更: {}", eventType);
                        break;
                    default:
                        break;
                }
            }, EVENT_EXECUTOR);
            
            treeCache.start();
            logger.info("注册根路径监听器成功: {}", rootPath);
        } catch (Exception e) {
            logger.error("注册根路径监听器失败", e);
        }
    }
    
    /**
     * 创建持久节点
     */
    public void createPersistentNode(String path, String data) {
        try {
            if (client.checkExists().forPath(path) == null) {
                client.create()
                    .creatingParentsIfNeeded()
                    .withMode(CreateMode.PERSISTENT)
                    .forPath(path, data.getBytes(StandardCharsets.UTF_8));
                logger.info("创建持久节点: {}", path);
            } else {
                updateNode(path, data);
            }
        } catch (Exception e) {
            logger.error("创建持久节点失败: {}", path, e);
        }
    }
    
    /**
     * 创建临时节点
     */
    public void createEphemeralNode(String path, String data) {
        try {
            if (client.checkExists().forPath(path) != null) {
                client.delete().forPath(path);
            }
            
            client.create()
                .creatingParentsIfNeeded()
                .withMode(CreateMode.EPHEMERAL)
                .forPath(path, data.getBytes(StandardCharsets.UTF_8));
            
            logger.info("创建临时节点: {}", path);
        } catch (Exception e) {
            logger.error("创建临时节点失败: {}", path, e);
        }
    }
    
    /**
     * 更新节点数据
     */
    public void updateNode(String path, String data) {
        try {
            client.setData().forPath(path, data.getBytes(StandardCharsets.UTF_8));
            logger.info("更新节点数据: {}", path);
        } catch (KeeperException.NoNodeException e) {
            logger.warn("节点不存在，尝试创建: {}", path);
            createPersistentNode(path, data);
        } catch (Exception e) {
            logger.error("更新节点数据失败: {}", path, e);
        }
    }
    
    /**
     * 删除节点
     */
    public void deleteNode(String path) {
        try {
            if (client.checkExists().forPath(path) != null) {
                client.delete()
                    .guaranteed()
                    .deletingChildrenIfNeeded()
                    .forPath(path);
                logger.info("删除节点: {}", path);
            }
        } catch (Exception e) {
            logger.error("删除节点失败: {}", path, e);
        }
    }
    
    /**
     * 获取节点数据
     */
    public String getNodeData(String path) {
        try {
            if (client.checkExists().forPath(path) != null) {
                byte[] data = client.getData().forPath(path);
                return new String(data, StandardCharsets.UTF_8);
            }
            return null;
        } catch (Exception e) {
            logger.error("获取节点数据失败: {}", path, e);
            return null;
        }
    }
    
    /**
     * 关闭客户端
     */
    public void close() {
        if (client != null && client.getState() == CuratorFrameworkState.STARTED) {
            client.close();
            logger.info("ZooKeeper客户端已关闭");
        }
        EVENT_EXECUTOR.shutdown();
    }
}

类职责说明

1. ZooKeeperConfigLoader

• 作用：从application.yml加载配置信息，提供类型安全的配置读取接口。
• 核心知识点：
  • YAML配置解析（使用SnakeYAML库）
  • 单例模式实现
  • 配置项的层级化管理
  • 类型转换与安全获取

2. ZooKeeperConfigManager

• 作用：管理配置缓存，处理配置变更事件，提供配置操作接口。
• 核心知识点：
  • 线程安全的配置缓存（ConcurrentHashMap）
  • 事件监听模式
  • 配置变更的广播机制
  • 配置加载与更新策略

3. ZookeeperClient

• 作用：与ZooKeeper服务器通信，实现节点操作和监听。
• 核心知识点：
  • CuratorFramework的使用
  • 连接管理与重试策略
  • 节点监听机制（TreeCache）
  • 持久化与临时节点的区别
  • 配置变更的实时响应

使用示例

// 获取配置值
String dbUrl = ZooKeeperConfigManager.getInstance().getConfig("database.url");

// 添加配置变更监听器
ZooKeeperConfigManager.getInstance().addConfigListener(new ConfigChangeListener() {
    @Override
    public void onConfigInserted(String key, String value) {
        System.out.println("配置插入: " + key + " = " + value);
    }
    
    @Override
    public void onConfigUpdated(String key, String oldValue, String newValue) {
        System.out.println("配置更新: " + key + " 从 " + oldValue + " 变为 " + newValue);
    }
    
    @Override
    public void onConfigDeleted(String key, String oldValue) {
        System.out.println("配置删除: " + key);
    }
});

// 创建ZooKeeper节点
ZookeeperClient.getInstance().createPersistentNode("/myapp/config/db.url", "jdbc:mysql://localhost:3306/mydb");