第298集RocketMQ集群搭建架构实战：分布式消息、高可用部署与企业级消息中间件解决方案

/**
 * RocketMQ集群核心组件
 */
@Component
public class RocketMQClusterManager {

    @Autowired
    private NameServerManager nameServerManager;

    @Autowired
    private BrokerManager brokerManager;

    @Autowired
    private TopicManager topicManager;

    @Autowired
    private ClusterMonitorService clusterMonitorService;

    @Autowired
    private SecurityManager securityManager;

    /**
     * 初始化RocketMQ集群
     */
    public void initializeCluster() {
        try {
            // 1. 启动NameServer集群
            startNameServerCluster();

            // 2. 启动Broker集群
            startBrokerCluster();

            // 3. 创建系统Topic
            createSystemTopics();

            // 4. 配置安全认证
            configureSecurity();

            // 5. 启动监控服务
            startMonitoringService();

            log.info("RocketMQ集群初始化完成");

        } catch (Exception e) {
            log.error("RocketMQ集群初始化失败", e);
            throw new ClusterInitializationException("集群初始化失败", e);
        }
    }

    /**
     * 启动RocketMQ集群
     */
    public void startCluster() {
        try {
            // 1. 检查集群状态
            checkClusterStatus();

            // 2. 启动所有NameServer
            nameServerManager.startAllNameServers();

            // 3. 启动所有Broker
            brokerManager.startAllBrokers();

            // 4. 验证集群健康状态
            validateClusterHealth();

            // 5. 启动监控
            clusterMonitorService.startMonitoring();

            log.info("RocketMQ集群启动成功");

        } catch (Exception e) {
            log.error("RocketMQ集群启动失败", e);
            throw new ClusterStartException("集群启动失败", e);
        }
    }

    /**
     * 停止RocketMQ集群
     */
    public void stopCluster() {
        try {
            // 1. 停止监控
            clusterMonitorService.stopMonitoring();

            // 2. 停止所有Broker
            brokerManager.stopAllBrokers();

            // 3. 停止所有NameServer
            nameServerManager.stopAllNameServers();

            log.info("RocketMQ集群停止成功");

        } catch (Exception e) {
            log.error("RocketMQ集群停止失败", e);
        }
    }

    /**
     * 启动NameServer集群
     */
    private void startNameServerCluster() {
        // 实现NameServer集群启动逻辑
        log.info("启动NameServer集群");
    }

    /**
     * 启动Broker集群
     */
    private void startBrokerCluster() {
        // 实现Broker集群启动逻辑
        log.info("启动Broker集群");
    }

    /**
     * 创建系统Topic
     */
    private void createSystemTopics() {
        // 实现系统Topic创建逻辑
        log.info("创建系统Topic");
    }

    /**
     * 配置安全认证
     */
    private void configureSecurity() {
        // 实现安全认证配置逻辑
        log.info("配置安全认证");
    }

    /**
     * 启动监控服务
     */
    private void startMonitoringService() {
        // 实现监控服务启动逻辑
        log.info("启动监控服务");
    }
}

/**
 * RocketMQ NameServer管理器
 */
@Service
public class NameServerManager {

    @Autowired
    private NameServerConfigService nameServerConfigService;

    @Autowired
    private NameServerDeploymentService nameServerDeploymentService;

    @Autowired
    private NameServerHealthChecker nameServerHealthChecker;

    private final Map<String, NameServerNode> nameServerNodes;

    public NameServerManager() {
        this.nameServerNodes = new ConcurrentHashMap<>();
    }

    /**
     * 启动所有NameServer
     */
    public void startAllNameServers() {
        try {
            // 1. 获取NameServer配置
            List<NameServerConfig> nameServerConfigs = nameServerConfigService.getAllNameServerConfigs();

            // 2. 启动每个NameServer
            for (NameServerConfig config : nameServerConfigs) {
                startNameServer(config);
            }

            // 3. 等待所有NameServer启动完成
            waitForNameServersToStart();

            // 4. 验证NameServer集群状态
            validateNameServerClusterState();

            log.info("所有NameServer启动完成，数量: {}", nameServerConfigs.size());

        } catch (Exception e) {
            log.error("NameServer启动失败", e);
            throw new NameServerStartException("NameServer启动失败", e);
        }
    }

    /**
     * 停止所有NameServer
     */
    public void stopAllNameServers() {
        try {
            // 1. 优雅关闭所有NameServer
            for (NameServerNode nameServerNode : nameServerNodes.values()) {
                stopNameServer(nameServerNode);
            }

            // 2. 等待所有NameServer停止完成
            waitForNameServersToStop();

            log.info("所有NameServer停止完成");

        } catch (Exception e) {
            log.error("NameServer停止失败", e);
        }
    }

    /**
     * 启动单个NameServer
     */
    private void startNameServer(NameServerConfig config) {
        try {
            // 1. 创建NameServer节点
            NameServerNode nameServerNode = createNameServerNode(config);

            // 2. 部署NameServer
            nameServerDeploymentService.deployNameServer(nameServerNode);

            // 3. 启动NameServer进程
            nameServerDeploymentService.startNameServerProcess(nameServerNode);

            // 4. 等待NameServer启动
            waitForNameServerToStart(nameServerNode);

            // 5. 验证NameServer健康状态
            validateNameServerHealth(nameServerNode);

            // 6. 注册NameServer节点
            nameServerNodes.put(nameServerNode.getNameServerId(), nameServerNode);

            log.info("NameServer启动成功: {}", nameServerNode.getNameServerId());

        } catch (Exception e) {
            log.error("NameServer启动失败: {}", config.getNameServerId(), e);
            throw new NameServerStartException("NameServer启动失败", e);
        }
    }

    /**
     * 停止单个NameServer
     */
    private void stopNameServer(NameServerNode nameServerNode) {
        try {
            // 1. 优雅关闭NameServer进程
            nameServerDeploymentService.stopNameServerProcess(nameServerNode);

            // 2. 等待NameServer停止
            waitForNameServerToStop(nameServerNode);

            // 3. 清理NameServer资源
            nameServerDeploymentService.cleanupNameServerResources(nameServerNode);

            // 4. 注销NameServer节点
            nameServerNodes.remove(nameServerNode.getNameServerId());

            log.info("NameServer停止成功: {}", nameServerNode.getNameServerId());

        } catch (Exception e) {
            log.error("NameServer停止失败: {}", nameServerNode.getNameServerId(), e);
        }
    }

    /**
     * 创建NameServer节点
     */
    private NameServerNode createNameServerNode(NameServerConfig config) {
        NameServerNode nameServerNode = new NameServerNode();
        nameServerNode.setNameServerId(config.getNameServerId());
        nameServerNode.setHost(config.getHost());
        nameServerNode.setPort(config.getPort());
        nameServerNode.setConfig(config);
        nameServerNode.setStatus(NameServerStatus.STARTING);
        nameServerNode.setStartTime(System.currentTimeMillis());

        return nameServerNode;
    }

    /**
     * 等待NameServer启动
     */
    private void waitForNameServerToStart(NameServerNode nameServerNode) {
        int maxWaitTime = 30000; // 30秒
        int checkInterval = 1000; // 1秒
        int waitedTime = 0;

        while (waitedTime < maxWaitTime) {
            try {
                if (nameServerHealthChecker.isNameServerHealthy(nameServerNode)) {
                    nameServerNode.setStatus(NameServerStatus.RUNNING);
                    return;
                }

                Thread.sleep(checkInterval);
                waitedTime += checkInterval;

            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new NameServerStartException("等待NameServer启动被中断", e);
            }
        }

        throw new NameServerStartException("NameServer启动超时: " + nameServerNode.getNameServerId());
    }

    /**
     * 等待NameServer停止
     */
    private void waitForNameServerToStop(NameServerNode nameServerNode) {
        int maxWaitTime = 30000; // 30秒
        int checkInterval = 1000; // 1秒
        int waitedTime = 0;

        while (waitedTime < maxWaitTime) {
            try {
                if (!nameServerHealthChecker.isNameServerRunning(nameServerNode)) {
                    nameServerNode.setStatus(NameServerStatus.STOPPED);
                    return;
                }

                Thread.sleep(checkInterval);
                waitedTime += checkInterval;

            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new NameServerStopException("等待NameServer停止被中断", e);
            }
        }

        throw new NameServerStopException("NameServer停止超时: " + nameServerNode.getNameServerId());
    }

    /**
     * 验证NameServer健康状态
     */
    private void validateNameServerHealth(NameServerNode nameServerNode) {
        try {
            NameServerHealthStatus healthStatus = nameServerHealthChecker.checkNameServerHealth(nameServerNode);

            if (!healthStatus.isHealthy()) {
                throw new NameServerHealthException("NameServer健康检查失败: " + nameServerNode.getNameServerId());
            }

            nameServerNode.setHealthStatus(healthStatus);

        } catch (Exception e) {
            log.error("NameServer健康检查失败: {}", nameServerNode.getNameServerId(), e);
            throw new NameServerHealthException("NameServer健康检查失败", e);
        }
    }

    /**
     * 等待所有NameServer启动完成
     */
    private void waitForNameServersToStart() {
        int maxWaitTime = 120000; // 2分钟
        int checkInterval = 5000; // 5秒
        int waitedTime = 0;

        while (waitedTime < maxWaitTime) {
            try {
                boolean allNameServersRunning = nameServerNodes.values().stream()
                    .allMatch(nameServer -> nameServer.getStatus() == NameServerStatus.RUNNING);

                if (allNameServersRunning) {
                    return;
                }

                Thread.sleep(checkInterval);
                waitedTime += checkInterval;

            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new NameServerStartException("等待所有NameServer启动被中断", e);
            }
        }

        throw new NameServerStartException("等待所有NameServer启动超时");
    }

    /**
     * 等待所有NameServer停止完成
     */
    private void waitForNameServersToStop() {
        int maxWaitTime = 120000; // 2分钟
        int checkInterval = 5000; // 5秒
        int waitedTime = 0;

        while (waitedTime < maxWaitTime) {
            try {
                boolean allNameServersStopped = nameServerNodes.values().stream()
                    .allMatch(nameServer -> nameServer.getStatus() == NameServerStatus.STOPPED);

                if (allNameServersStopped) {
                    return;
                }

                Thread.sleep(checkInterval);
                waitedTime += checkInterval;

            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new NameServerStopException("等待所有NameServer停止被中断", e);
            }
        }

        throw new NameServerStopException("等待所有NameServer停止超时");
    }

    /**
     * 验证NameServer集群状态
     */
    private void validateNameServerClusterState() {
        try {
            // 检查集群是否形成
            if (nameServerNodes.size() < 2) {
                log.warn("NameServer集群节点数量不足，建议至少2个节点");
            }

            // 检查所有NameServer是否健康
            for (NameServerNode nameServerNode : nameServerNodes.values()) {
                if (nameServerNode.getHealthStatus() == null || !nameServerNode.getHealthStatus().isHealthy()) {
                    throw new NameServerClusterValidationException("NameServer集群状态验证失败，NameServer不健康: " + nameServerNode.getNameServerId());
                }
            }

            log.info("NameServer集群状态验证通过");

        } catch (Exception e) {
            log.error("NameServer集群状态验证失败", e);
            throw new NameServerClusterValidationException("NameServer集群状态验证失败", e);
        }
    }
}

/**
 * RocketMQ NameServer配置服务
 */
@Service
public class NameServerConfigService {

    @Autowired
    private ConfigurationRepository configRepository;

    /**
     * 获取所有NameServer配置
     */
    public List<NameServerConfig> getAllNameServerConfigs() {
        return configRepository.findAllNameServerConfigs();
    }

    /**
     * 获取NameServer配置
     */
    public NameServerConfig getNameServerConfig(String nameServerId) {
        return configRepository.findNameServerConfigById(nameServerId)
            .orElseThrow(() -> new ConfigNotFoundException("NameServer配置不存在: " + nameServerId));
    }

    /**
     * 保存NameServer配置
     */
    public void saveNameServerConfig(NameServerConfig config) {
        try {
            // 验证配置
            validateNameServerConfig(config);

            // 保存配置
            configRepository.saveNameServerConfig(config);

            log.info("NameServer配置保存成功: {}", config.getNameServerId());

        } catch (Exception e) {
            log.error("NameServer配置保存失败", e);
            throw new ConfigSaveException("NameServer配置保存失败", e);
        }
    }

    /**
     * 更新NameServer配置
     */
    public void updateNameServerConfig(String nameServerId, NameServerConfig config) {
        try {
            // 检查配置是否存在
            if (!configRepository.existsNameServerConfig(nameServerId)) {
                throw new ConfigNotFoundException("NameServer配置不存在: " + nameServerId);
            }

            // 验证配置
            validateNameServerConfig(config);

            // 更新配置
            config.setNameServerId(nameServerId);
            configRepository.saveNameServerConfig(config);

            log.info("NameServer配置更新成功: {}", nameServerId);

        } catch (Exception e) {
            log.error("NameServer配置更新失败", e);
            throw new ConfigUpdateException("NameServer配置更新失败", e);
        }
    }

    /**
     * 删除NameServer配置
     */
    public void deleteNameServerConfig(String nameServerId) {
        try {
            if (!configRepository.existsNameServerConfig(nameServerId)) {
                throw new ConfigNotFoundException("NameServer配置不存在: " + nameServerId);
            }

            configRepository.deleteNameServerConfig(nameServerId);

            log.info("NameServer配置删除成功: {}", nameServerId);

        } catch (Exception e) {
            log.error("NameServer配置删除失败", e);
            throw new ConfigDeleteException("NameServer配置删除失败", e);
        }
    }

    /**
     * 生成NameServer配置文件
     */
    public String generateNameServerConfigFile(NameServerConfig config) {
        StringBuilder configContent = new StringBuilder();

        // 基础配置
        configContent.append("# RocketMQ NameServer Configuration\n");
        configContent.append("listenPort=").append(config.getPort()).append("\n");
        configContent.append("serverWorkerThreads=").append(config.getServerWorkerThreads()).append("\n");
        configContent.append("serverCallbackExecutorThreads=").append(config.getServerCallbackExecutorThreads()).append("\n");
        configContent.append("serverSelectorThreads=").append(config.getServerSelectorThreads()).append("\n");
        configContent.append("serverOnewaySemaphoreValue=").append(config.getServerOnewaySemaphoreValue()).append("\n");
        configContent.append("serverAsyncSemaphoreValue=").append(config.getServerAsyncSemaphoreValue()).append("\n");

        // 网络配置
        configContent.append("serverChannelMaxIdleTimeSeconds=").append(config.getServerChannelMaxIdleTimeSeconds()).append("\n");
        configContent.append("serverSocketSndBufSize=").append(config.getServerSocketSndBufSize()).append("\n");
        configContent.append("serverSocketRcvBufSize=").append(config.getServerSocketRcvBufSize()).append("\n");
        configContent.append("serverPooledByteBufAllocatorEnable=").append(config.isServerPooledByteBufAllocatorEnable()).append("\n");

        // 日志配置
        configContent.append("rocketmqHome=").append(config.getRocketmqHome()).append("\n");
        configContent.append("namesrvAddr=").append(config.getNamesrvAddr()).append("\n");
        configContent.append("rocketmqHome=").append(config.getRocketmqHome()).append("\n");
        configContent.append("kvConfigPath=").append(config.getKvConfigPath()).append("\n");
        configContent.append("configStorePath=").append(config.getConfigStorePath()).append("\n");

        // 性能配置
        configContent.append("serverChannelMaxIdleTimeSeconds=").append(config.getServerChannelMaxIdleTimeSeconds()).append("\n");
        configContent.append("serverSocketSndBufSize=").append(config.getServerSocketSndBufSize()).append("\n");
        configContent.append("serverSocketRcvBufSize=").append(config.getServerSocketRcvBufSize()).append("\n");

        return configContent.toString();
    }

    /**
     * 验证NameServer配置
     */
    private void validateNameServerConfig(NameServerConfig config) {
        if (config.getNameServerId() == null || config.getNameServerId().isEmpty()) {
            throw new ConfigValidationException("NameServer ID不能为空");
        }

        if (config.getHost() == null || config.getHost().isEmpty()) {
            throw new ConfigValidationException("NameServer主机地址不能为空");
        }

        if (config.getPort() <= 0 || config.getPort() > 65535) {
            throw new ConfigValidationException("NameServer端口必须在1-65535之间");
        }

        if (config.getRocketmqHome() == null || config.getRocketmqHome().isEmpty()) {
            throw new ConfigValidationException("RocketMQ Home目录不能为空");
        }

        if (config.getKvConfigPath() == null || config.getKvConfigPath().isEmpty()) {
            throw new ConfigValidationException("KV配置路径不能为空");
        }

        if (config.getConfigStorePath() == null || config.getConfigStorePath().isEmpty()) {
            throw new ConfigValidationException("配置存储路径不能为空");
        }
    }
}

/**
 * RocketMQ Broker管理器
 */
@Service
public class BrokerManager {

    @Autowired
    private BrokerConfigService brokerConfigService;

    @Autowired
    private BrokerDeploymentService brokerDeploymentService;

    @Autowired
    private BrokerHealthChecker brokerHealthChecker;

    private final Map<String, BrokerNode> brokerNodes;

    public BrokerManager() {
        this.brokerNodes = new ConcurrentHashMap<>();
    }

    /**
     * 启动所有Broker
     */
    public void startAllBrokers() {
        try {
            // 1. 获取Broker配置
            List<BrokerConfig> brokerConfigs = brokerConfigService.getAllBrokerConfigs();

            // 2. 启动每个Broker
            for (BrokerConfig config : brokerConfigs) {
                startBroker(config);
            }

            // 3. 等待所有Broker启动完成
            waitForBrokersToStart();

            // 4. 验证Broker集群状态
            validateBrokerClusterState();

            log.info("所有Broker启动完成，数量: {}", brokerConfigs.size());

        } catch (Exception e) {
            log.error("Broker启动失败", e);
            throw new BrokerStartException("Broker启动失败", e);
        }
    }

    /**
     * 停止所有Broker
     */
    public void stopAllBrokers() {
        try {
            // 1. 优雅关闭所有Broker
            for (BrokerNode brokerNode : brokerNodes.values()) {
                stopBroker(brokerNode);
            }

            // 2. 等待所有Broker停止完成
            waitForBrokersToStop();

            log.info("所有Broker停止完成");

        } catch (Exception e) {
            log.error("Broker停止失败", e);
        }
    }

    /**
     * 启动单个Broker
     */
    private void startBroker(BrokerConfig config) {
        try {
            // 1. 创建Broker节点
            BrokerNode brokerNode = createBrokerNode(config);

            // 2. 部署Broker
            brokerDeploymentService.deployBroker(brokerNode);

            // 3. 启动Broker进程
            brokerDeploymentService.startBrokerProcess(brokerNode);

            // 4. 等待Broker启动
            waitForBrokerToStart(brokerNode);

            // 5. 验证Broker健康状态
            validateBrokerHealth(brokerNode);

            // 6. 注册Broker节点
            brokerNodes.put(brokerNode.getBrokerId(), brokerNode);

            log.info("Broker启动成功: {}", brokerNode.getBrokerId());

        } catch (Exception e) {
            log.error("Broker启动失败: {}", config.getBrokerId(), e);
            throw new BrokerStartException("Broker启动失败", e);
        }
    }

    /**
     * 停止单个Broker
     */
    private void stopBroker(BrokerNode brokerNode) {
        try {
            // 1. 优雅关闭Broker进程
            brokerDeploymentService.stopBrokerProcess(brokerNode);

            // 2. 等待Broker停止
            waitForBrokerToStop(brokerNode);

            // 3. 清理Broker资源
            brokerDeploymentService.cleanupBrokerResources(brokerNode);

            // 4. 注销Broker节点
            brokerNodes.remove(brokerNode.getBrokerId());

            log.info("Broker停止成功: {}", brokerNode.getBrokerId());

        } catch (Exception e) {
            log.error("Broker停止失败: {}", brokerNode.getBrokerId(), e);
        }
    }

    /**
     * 创建Broker节点
     */
    private BrokerNode createBrokerNode(BrokerConfig config) {
        BrokerNode brokerNode = new BrokerNode();
        brokerNode.setBrokerId(config.getBrokerId());
        brokerNode.setBrokerName(config.getBrokerName());
        brokerNode.setHost(config.getHost());
        brokerNode.setPort(config.getPort());
        brokerNode.setBrokerRole(config.getBrokerRole());
        brokerNode.setBrokerClusterName(config.getBrokerClusterName());
        brokerNode.setConfig(config);
        brokerNode.setStatus(BrokerStatus.STARTING);
        brokerNode.setStartTime(System.currentTimeMillis());

        return brokerNode;
    }

    /**
     * 等待Broker启动
     */
    private void waitForBrokerToStart(BrokerNode brokerNode) {
        int maxWaitTime = 60000; // 60秒
        int checkInterval = 2000; // 2秒
        int waitedTime = 0;

        while (waitedTime < maxWaitTime) {
            try {
                if (brokerHealthChecker.isBrokerHealthy(brokerNode)) {
                    brokerNode.setStatus(BrokerStatus.RUNNING);
                    return;
                }

                Thread.sleep(checkInterval);
                waitedTime += checkInterval;

            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new BrokerStartException("等待Broker启动被中断", e);
            }
        }

        throw new BrokerStartException("Broker启动超时: " + brokerNode.getBrokerId());
    }

    /**
     * 等待Broker停止
     */
    private void waitForBrokerToStop(BrokerNode brokerNode) {
        int maxWaitTime = 60000; // 60秒
        int checkInterval = 2000; // 2秒
        int waitedTime = 0;

        while (waitedTime < maxWaitTime) {
            try {
                if (!brokerHealthChecker.isBrokerRunning(brokerNode)) {
                    brokerNode.setStatus(BrokerStatus.STOPPED);
                    return;
                }

                Thread.sleep(checkInterval);
                waitedTime += checkInterval;

            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new BrokerStopException("等待Broker停止被中断", e);
            }
        }

        throw new BrokerStopException("Broker停止超时: " + brokerNode.getBrokerId());
    }

    /**
     * 验证Broker健康状态
     */
    private void validateBrokerHealth(BrokerNode brokerNode) {
        try {
            BrokerHealthStatus healthStatus = brokerHealthChecker.checkBrokerHealth(brokerNode);

            if (!healthStatus.isHealthy()) {
                throw new BrokerHealthException("Broker健康检查失败: " + brokerNode.getBrokerId());
            }

            brokerNode.setHealthStatus(healthStatus);

        } catch (Exception e) {
            log.error("Broker健康检查失败: {}", brokerNode.getBrokerId(), e);
            throw new BrokerHealthException("Broker健康检查失败", e);
        }
    }

    /**
     * 等待所有Broker启动完成
     */
    private void waitForBrokersToStart() {
        int maxWaitTime = 300000; // 5分钟
        int checkInterval = 10000; // 10秒
        int waitedTime = 0;

        while (waitedTime < maxWaitTime) {
            try {
                boolean allBrokersRunning = brokerNodes.values().stream()
                    .allMatch(broker -> broker.getStatus() == BrokerStatus.RUNNING);

                if (allBrokersRunning) {
                    return;
                }

                Thread.sleep(checkInterval);
                waitedTime += checkInterval;

            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new BrokerStartException("等待所有Broker启动被中断", e);
            }
        }

        throw new BrokerStartException("等待所有Broker启动超时");
    }

    /**
     * 等待所有Broker停止完成
     */
    private void waitForBrokersToStop() {
        int maxWaitTime = 300000; // 5分钟
        int checkInterval = 10000; // 10秒
        int waitedTime = 0;

        while (waitedTime < maxWaitTime) {
            try {
                boolean allBrokersStopped = brokerNodes.values().stream()
                    .allMatch(broker -> broker.getStatus() == BrokerStatus.STOPPED);

                if (allBrokersStopped) {
                    return;
                }

                Thread.sleep(checkInterval);
                waitedTime += checkInterval;

            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new BrokerStopException("等待所有Broker停止被中断", e);
            }
        }

        throw new BrokerStopException("等待所有Broker停止超时");
    }

    /**
     * 验证Broker集群状态
     */
    private void validateBrokerClusterState() {
        try {
            // 检查集群是否形成
            if (brokerNodes.size() < 2) {
                log.warn("Broker集群节点数量不足，建议至少2个节点");
            }

            // 检查所有Broker是否健康
            for (BrokerNode brokerNode : brokerNodes.values()) {
                if (brokerNode.getHealthStatus() == null || !brokerNode.getHealthStatus().isHealthy()) {
                    throw new BrokerClusterValidationException("Broker集群状态验证失败，Broker不健康: " + brokerNode.getBrokerId());
                }
            }

            log.info("Broker集群状态验证通过");

        } catch (Exception e) {
            log.error("Broker集群状态验证失败", e);
            throw new BrokerClusterValidationException("Broker集群状态验证失败", e);
        }
    }
}

/**
 * RocketMQ Broker配置服务
 */
@Service
public class BrokerConfigService {

    @Autowired
    private ConfigurationRepository configRepository;

    /**
     * 获取所有Broker配置
     */
    public List<BrokerConfig> getAllBrokerConfigs() {
        return configRepository.findAllBrokerConfigs();
    }

    /**
     * 获取Broker配置
     */
    public BrokerConfig getBrokerConfig(String brokerId) {
        return configRepository.findBrokerConfigById(brokerId)
            .orElseThrow(() -> new ConfigNotFoundException("Broker配置不存在: " + brokerId));
    }

    /**
     * 保存Broker配置
     */
    public void saveBrokerConfig(BrokerConfig config) {
        try {
            // 验证配置
            validateBrokerConfig(config);

            // 保存配置
            configRepository.saveBrokerConfig(config);

            log.info("Broker配置保存成功: {}", config.getBrokerId());

        } catch (Exception e) {
            log.error("Broker配置保存失败", e);
            throw new ConfigSaveException("Broker配置保存失败", e);
        }
    }

    /**
     * 更新Broker配置
     */
    public void updateBrokerConfig(String brokerId, BrokerConfig config) {
        try {
            // 检查配置是否存在
            if (!configRepository.existsBrokerConfig(brokerId)) {
                throw new ConfigNotFoundException("Broker配置不存在: " + brokerId);
            }

            // 验证配置
            validateBrokerConfig(config);

            // 更新配置
            config.setBrokerId(brokerId);
            configRepository.saveBrokerConfig(config);

            log.info("Broker配置更新成功: {}", brokerId);

        } catch (Exception e) {
            log.error("Broker配置更新失败", e);
            throw new ConfigUpdateException("Broker配置更新失败", e);
        }
    }

    /**
     * 删除Broker配置
     */
    public void deleteBrokerConfig(String brokerId) {
        try {
            if (!configRepository.existsBrokerConfig(brokerId)) {
                throw new ConfigNotFoundException("Broker配置不存在: " + brokerId);
            }

            configRepository.deleteBrokerConfig(brokerId);

            log.info("Broker配置删除成功: {}", brokerId);

        } catch (Exception e) {
            log.error("Broker配置删除失败", e);
            throw new ConfigDeleteException("Broker配置删除失败", e);
        }
    }

    /**
     * 生成Broker配置文件
     */
    public String generateBrokerConfigFile(BrokerConfig config) {
        StringBuilder configContent = new StringBuilder();

        // 基础配置
        configContent.append("# RocketMQ Broker Configuration\n");
        configContent.append("brokerClusterName=").append(config.getBrokerClusterName()).append("\n");
        configContent.append("brokerName=").append(config.getBrokerName()).append("\n");
        configContent.append("brokerId=").append(config.getBrokerId()).append("\n");
        configContent.append("brokerRole=").append(config.getBrokerRole()).append("\n");
        configContent.append("namesrvAddr=").append(config.getNamesrvAddr()).append("\n");
        configContent.append("listenPort=").append(config.getListenPort()).append("\n");

        // 存储配置
        configContent.append("storePathRootDir=").append(config.getStorePathRootDir()).append("\n");
        configContent.append("storePathCommitLog=").append(config.getStorePathCommitLog()).append("\n");
        configContent.append("storePathConsumeQueue=").append(config.getStorePathConsumeQueue()).append("\n");
        configContent.append("storePathIndex=").append(config.getStorePathIndex()).append("\n");
        configContent.append("storePathDLedger=").append(config.getStorePathDLedger()).append("\n");

        // 消息存储配置
        configContent.append("mapedFileSizeCommitLog=").append(config.getMapedFileSizeCommitLog()).append("\n");
        configContent.append("mapedFileSizeConsumeQueue=").append(config.getMapedFileSizeConsumeQueue()).append("\n");
        configContent.append("maxTransferBytesOnMessageInMemory=").append(config.getMaxTransferBytesOnMessageInMemory()).append("\n");
        configContent.append("maxTransferCountOnMessageInMemory=").append(config.getMaxTransferCountOnMessageInMemory()).append("\n");
        configContent.append("maxTransferBytesOnMessageInDisk=").append(config.getMaxTransferBytesOnMessageInDisk()).append("\n");
        configContent.append("maxTransferCountOnMessageInDisk=").append(config.getMaxTransferCountOnMessageInDisk()).append("\n");

        // 消息过期配置
        configContent.append("messageDelayLevel=").append(config.getMessageDelayLevel()).append("\n");
        configContent.append("messageMaxSize=").append(config.getMessageMaxSize()).append("\n");
        configContent.append("messageMaxSize=").append(config.getMessageMaxSize()).append("\n");

        // 刷盘配置
        configContent.append("flushDiskType=").append(config.getFlushDiskType()).append("\n");
        configContent.append("flushCommitLogLeastPages=").append(config.getFlushCommitLogLeastPages()).append("\n");
        configContent.append("flushCommitLogThoroughInterval=").append(config.getFlushCommitLogThoroughInterval()).append("\n");
        configContent.append("flushConsumeQueueLeastPages=").append(config.getFlushConsumeQueueLeastPages()).append("\n");
        configContent.append("flushConsumeQueueThoroughInterval=").append(config.getFlushConsumeQueueThoroughInterval()).append("\n");

        // 网络配置
        configContent.append("listenPort=").append(config.getListenPort()).append("\n");
        configContent.append("serverWorkerThreads=").append(config.getServerWorkerThreads()).append("\n");
        configContent.append("serverCallbackExecutorThreads=").append(config.getServerCallbackExecutorThreads()).append("\n");
        configContent.append("serverSelectorThreads=").append(config.getServerSelectorThreads()).append("\n");
        configContent.append("serverOnewaySemaphoreValue=").append(config.getServerOnewaySemaphoreValue()).append("\n");
        configContent.append("serverAsyncSemaphoreValue=").append(config.getServerAsyncSemaphoreValue()).append("\n");

        // 性能配置
        configContent.append("sendMessageThreadPoolNums=").append(config.getSendMessageThreadPoolNums()).append("\n");
        configContent.append("pullMessageThreadPoolNums=").append(config.getPullMessageThreadPoolNums()).append("\n");
        configContent.append("queryMessageThreadPoolNums=").append(config.getQueryMessageThreadPoolNums()).append("\n");
        configContent.append("adminBrokerThreadPoolNums=").append(config.getAdminBrokerThreadPoolNums()).append("\n");
        configContent.append("clientManageThreadPoolNums=").append(config.getClientManageThreadPoolNums()).append("\n");
        configContent.append("consumerManageThreadPoolNums=").append(config.getConsumerManageThreadPoolNums()).append("\n");

        return configContent.toString();
    }

    /**
     * 验证Broker配置
     */
    private void validateBrokerConfig(BrokerConfig config) {
        if (config.getBrokerId() == null || config.getBrokerId().isEmpty()) {
            throw new ConfigValidationException("Broker ID不能为空");
        }

        if (config.getBrokerName() == null || config.getBrokerName().isEmpty()) {
            throw new ConfigValidationException("Broker名称不能为空");
        }

        if (config.getBrokerClusterName() == null || config.getBrokerClusterName().isEmpty()) {
            throw new ConfigValidationException("Broker集群名称不能为空");
        }

        if (config.getHost() == null || config.getHost().isEmpty()) {
            throw new ConfigValidationException("Broker主机地址不能为空");
        }

        if (config.getListenPort() <= 0 || config.getListenPort() > 65535) {
            throw new ConfigValidationException("Broker监听端口必须在1-65535之间");
        }

        if (config.getNamesrvAddr() == null || config.getNamesrvAddr().isEmpty()) {
            throw new ConfigValidationException("NameServer地址不能为空");
        }

        if (config.getStorePathRootDir() == null || config.getStorePathRootDir().isEmpty()) {
            throw new ConfigValidationException("存储根目录不能为空");
        }

        if (config.getBrokerRole() == null) {
            throw new ConfigValidationException("Broker角色不能为空");
        }

        // 验证Broker角色
        if (!isValidBrokerRole(config.getBrokerRole())) {
            throw new ConfigValidationException("Broker角色无效: " + config.getBrokerRole());
        }
    }

    /**
     * 验证Broker角色
     */
    private boolean isValidBrokerRole(String brokerRole) {
        return "ASYNC_MASTER".equals(brokerRole) || 
               "SYNC_MASTER".equals(brokerRole) || 
               "SLAVE".equals(brokerRole);
    }
}

/**
 * RocketMQ Topic管理器
 */
@Service
public class TopicManager {

    @Autowired
    private RocketMQAdminClient rocketMQAdminClient;

    @Autowired
    private TopicConfigService topicConfigService;

    @Autowired
    private QueueManager queueManager;

    /**
     * 创建Topic
     */
    public void createTopic(TopicConfig topicConfig) {
        try {
            // 1. 验证Topic配置
            validateTopicConfig(topicConfig);

            // 2. 检查Topic是否已存在
            if (topicExists(topicConfig.getTopicName())) {
                throw new TopicExistsException("Topic已存在: " + topicConfig.getTopicName());
            }

            // 3. 创建Topic
            CreateTopicRequest request = new CreateTopicRequest();
            request.setTopic(topicConfig.getTopicName());
            request.setDefaultTopicQueueNums(topicConfig.getDefaultTopicQueueNums());
            request.setPerm(topicConfig.getPerm());
            request.setTopicFilterType(topicConfig.getTopicFilterType());
            request.setTopicSysFlag(topicConfig.getTopicSysFlag());
            request.setOrder(topicConfig.isOrder());

            // 4. 执行创建
            rocketMQAdminClient.createTopic(request);

            // 5. 保存Topic配置
            topicConfigService.saveTopicConfig(topicConfig);

            log.info("Topic创建成功: {}", topicConfig.getTopicName());

        } catch (Exception e) {
            log.error("Topic创建失败: {}", topicConfig.getTopicName(), e);
            throw new TopicCreationException("Topic创建失败", e);
        }
    }

    /**
     * 删除Topic
     */
    public void deleteTopic(String topicName) {
        try {
            // 1. 检查Topic是否存在
            if (!topicExists(topicName)) {
                throw new TopicNotFoundException("Topic不存在: " + topicName);
            }

            // 2. 删除Topic
            DeleteTopicRequest request = new DeleteTopicRequest();
            request.setTopic(topicName);

            rocketMQAdminClient.deleteTopic(request);

            // 3. 删除Topic配置
            topicConfigService.deleteTopicConfig(topicName);

            log.info("Topic删除成功: {}", topicName);

        } catch (Exception e) {
            log.error("Topic删除失败: {}", topicName, e);
            throw new TopicDeletionException("Topic删除失败", e);
        }
    }

    /**
     * 更新Topic配置
     */
    public void updateTopicConfig(String topicName, TopicConfig topicConfig) {
        try {
            // 1. 检查Topic是否存在
            if (!topicExists(topicName)) {
                throw new TopicNotFoundException("Topic不存在: " + topicName);
            }

            // 2. 更新Topic配置
            UpdateTopicRequest request = new UpdateTopicRequest();
            request.setTopic(topicName);
            request.setDefaultTopicQueueNums(topicConfig.getDefaultTopicQueueNums());
            request.setPerm(topicConfig.getPerm());
            request.setTopicFilterType(topicConfig.getTopicFilterType());
            request.setTopicSysFlag(topicConfig.getTopicSysFlag());
            request.setOrder(topicConfig.isOrder());

            rocketMQAdminClient.updateTopic(request);

            // 3. 更新Topic配置
            topicConfigService.updateTopicConfig(topicName, topicConfig);

            log.info("Topic配置更新成功: {}", topicName);

        } catch (Exception e) {
            log.error("Topic配置更新失败: {}", topicName, e);
            throw new TopicConfigUpdateException("Topic配置更新失败", e);
        }
    }

    /**
     * 获取Topic信息
     */
    public TopicInfo getTopicInfo(String topicName) {
        try {
            // 1. 检查Topic是否存在
            if (!topicExists(topicName)) {
                throw new TopicNotFoundException("Topic不存在: " + topicName);
            }

            // 2. 获取Topic详细信息
            TopicRouteInfo topicRouteInfo = rocketMQAdminClient.getTopicRouteInfo(topicName);

            // 3. 构建Topic信息
            TopicInfo topicInfo = new TopicInfo();
            topicInfo.setTopicName(topicName);
            topicInfo.setQueueDatas(topicRouteInfo.getQueueDatas());
            topicInfo.setBrokerDatas(topicRouteInfo.getBrokerDatas());
            topicInfo.setFilterServerTable(topicRouteInfo.getFilterServerTable());

            return topicInfo;

        } catch (Exception e) {
            log.error("获取Topic信息失败: {}", topicName, e);
            throw new TopicInfoException("获取Topic信息失败", e);
        }
    }

    /**
     * 获取所有Topic列表
     */
    public List<TopicInfo> getAllTopics() {
        try {
            // 1. 获取所有Topic名称
            List<String> topicNames = rocketMQAdminClient.getAllTopicNames();

            // 2. 获取每个Topic的详细信息
            List<TopicInfo> topicInfos = new ArrayList<>();
            for (String topicName : topicNames) {
                try {
                    TopicInfo topicInfo = getTopicInfo(topicName);
                    topicInfos.add(topicInfo);
                } catch (Exception e) {
                    log.warn("获取Topic信息失败: {}", topicName, e);
                }
            }

            return topicInfos;

        } catch (Exception e) {
            log.error("获取所有Topic列表失败", e);
            throw new TopicListException("获取所有Topic列表失败", e);
        }
    }

    /**
     * 检查Topic是否存在
     */
    public boolean topicExists(String topicName) {
        try {
            List<String> topicNames = rocketMQAdminClient.getAllTopicNames();
            return topicNames.contains(topicName);

        } catch (Exception e) {
            log.error("检查Topic是否存在失败: {}", topicName, e);
            return false;
        }
    }

    /**
     * 验证Topic配置
     */
    private void validateTopicConfig(TopicConfig topicConfig) {
        if (topicConfig.getTopicName() == null || topicConfig.getTopicName().isEmpty()) {
            throw new TopicConfigValidationException("Topic名称不能为空");
        }

        if (topicConfig.getDefaultTopicQueueNums() <= 0) {
            throw new TopicConfigValidationException("默认队列数必须大于0");
        }

        if (topicConfig.getPerm() == null) {
            throw new TopicConfigValidationException("Topic权限不能为空");
        }

        // 验证Topic名称格式
        if (!isValidTopicName(topicConfig.getTopicName())) {
            throw new TopicConfigValidationException("Topic名称格式不正确");
        }

        // 验证权限
        if (!isValidPerm(topicConfig.getPerm())) {
            throw new TopicConfigValidationException("Topic权限无效: " + topicConfig.getPerm());
        }
    }

    /**
     * 验证Topic名称格式
     */
    private boolean isValidTopicName(String topicName) {
        // Topic名称只能包含字母、数字、点、下划线和连字符
        return topicName.matches("^[a-zA-Z0-9._-]+$");
    }

    /**
     * 验证权限
     */
    private boolean isValidPerm(int perm) {
        // 权限值：2=读，4=写，6=读写
        return perm == 2 || perm == 4 || perm == 6;
    }
}

/**
 * RocketMQ队列管理器
 */
@Service
public class QueueManager {

    @Autowired
    private RocketMQAdminClient rocketMQAdminClient;

    @Autowired
    private BrokerManager brokerManager;

    /**
     * 增加Topic队列数
     */
    public void increaseQueueNums(String topicName, int newQueueNums) {
        try {
            // 1. 检查Topic是否存在
            if (!topicExists(topicName)) {
                throw new TopicNotFoundException("Topic不存在: " + topicName);
            }

            // 2. 获取当前队列数
            int currentQueueNums = getCurrentQueueNums(topicName);

            // 3. 验证新队列数
            if (newQueueNums <= currentQueueNums) {
                throw new QueueValidationException("新队列数必须大于当前队列数");
            }

            // 4. 更新Topic队列数
            UpdateTopicRequest request = new UpdateTopicRequest();
            request.setTopic(topicName);
            request.setDefaultTopicQueueNums(newQueueNums);

            rocketMQAdminClient.updateTopic(request);

            log.info("Topic队列数增加成功: {} -> {}", topicName, newQueueNums);

        } catch (Exception e) {
            log.error("增加Topic队列数失败: {}", topicName, e);
            throw new QueueIncreaseException("增加Topic队列数失败", e);
        }
    }

    /**
     * 重新分配队列
     */
    public void reassignQueues(String topicName, Map<String, List<Integer>> brokerQueues) {
        try {
            // 1. 检查Topic是否存在
            if (!topicExists(topicName)) {
                throw new TopicNotFoundException("Topic不存在: " + topicName);
            }

            // 2. 验证队列分配
            validateQueueAssignment(topicName, brokerQueues);

            // 3. 创建队列重分配请求
            ReassignQueueRequest request = new ReassignQueueRequest();
            request.setTopic(topicName);
            request.setBrokerQueues(brokerQueues);

            // 4. 执行队列重分配
            rocketMQAdminClient.reassignQueue(request);

            log.info("Topic队列重分配成功: {}", topicName);

        } catch (Exception e) {
            log.error("队列重分配失败: {}", topicName, e);
            throw new QueueReassignmentException("队列重分配失败", e);
        }
    }

    /**
     * 获取队列信息
     */
    public List<QueueInfo> getQueueInfo(String topicName) {
        try {
            // 1. 检查Topic是否存在
            if (!topicExists(topicName)) {
                throw new TopicNotFoundException("Topic不存在: " + topicName);
            }

            // 2. 获取Topic路由信息
            TopicRouteInfo topicRouteInfo = rocketMQAdminClient.getTopicRouteInfo(topicName);

            // 3. 构建队列信息
            List<QueueInfo> queueInfos = new ArrayList<>();
            for (QueueData queueData : topicRouteInfo.getQueueDatas()) {
                QueueInfo queueInfo = new QueueInfo();
                queueInfo.setTopicName(topicName);
                queueInfo.setBrokerName(queueData.getBrokerName());
                queueInfo.setReadQueueNums(queueData.getReadQueueNums());
                queueInfo.setWriteQueueNums(queueData.getWriteQueueNums());
                queueInfo.setPerm(queueData.getPerm());
                queueInfo.setTopicSysFlag(queueData.getTopicSysFlag());

                queueInfos.add(queueInfo);
            }

            return queueInfos;

        } catch (Exception e) {
            log.error("获取队列信息失败: {}", topicName, e);
            throw new QueueInfoException("获取队列信息失败", e);
        }
    }

    /**
     * 检查队列健康状态
     */
    public QueueHealthStatus checkQueueHealth(String topicName) {
        try {
            // 1. 获取队列信息
            List<QueueInfo> queueInfos = getQueueInfo(topicName);

            // 2. 分析队列健康状态
            QueueHealthStatus healthStatus = new QueueHealthStatus();
            healthStatus.setTopicName(topicName);
            healthStatus.setTotalQueues(queueInfos.size());

            int healthyQueues = 0;
            int unbalancedQueues = 0;
            int readonlyQueues = 0;

            for (QueueInfo queueInfo : queueInfos) {
                // 检查读写队列数
                int readQueueNums = queueInfo.getReadQueueNums();
                int writeQueueNums = queueInfo.getWriteQueueNums();

                if (readQueueNums != writeQueueNums) {
                    unbalancedQueues++;
                }

                // 检查权限
                int perm = queueInfo.getPerm();
                if (perm == 2) { // 只读
                    readonlyQueues++;
                }

                // 检查队列是否健康
                if (readQueueNums == writeQueueNums && perm == 6) { // 读写
                    healthyQueues++;
                }
            }

            healthStatus.setHealthyQueues(healthyQueues);
            healthStatus.setUnbalancedQueues(unbalancedQueues);
            healthStatus.setReadonlyQueues(readonlyQueues);

            // 计算健康率
            double healthRate = (double) healthyQueues / queueInfos.size();
            healthStatus.setHealthRate(healthRate);

            // 判断整体健康状态
            healthStatus.setHealthy(healthRate >= 0.8); // 80%以上队列健康

            return healthStatus;

        } catch (Exception e) {
            log.error("检查队列健康状态失败: {}", topicName, e);
            throw new QueueHealthException("检查队列健康状态失败", e);
        }
    }

    /**
     * 获取当前队列数
     */
    private int getCurrentQueueNums(String topicName) {
        try {
            TopicRouteInfo topicRouteInfo = rocketMQAdminClient.getTopicRouteInfo(topicName);
            return topicRouteInfo.getQueueDatas().stream()
                .mapToInt(QueueData::getWriteQueueNums)
                .sum();

        } catch (Exception e) {
            log.error("获取当前队列数失败: {}", topicName, e);
            throw new QueueInfoException("获取当前队列数失败", e);
        }
    }

    /**
     * 验证队列分配
     */
    private void validateQueueAssignment(String topicName, Map<String, List<Integer>> brokerQueues) {
        // 1. 检查Broker是否存在
        for (String brokerName : brokerQueues.keySet()) {
            if (!brokerManager.isBrokerExists(brokerName)) {
                throw new QueueValidationException("Broker不存在: " + brokerName);
            }
        }

        // 2. 检查队列数
        for (Map.Entry<String, List<Integer>> entry : brokerQueues.entrySet()) {
            List<Integer> queues = entry.getValue();

            if (queues.isEmpty()) {
                throw new QueueValidationException("Broker队列不能为空: " + entry.getKey());
            }

            // 检查队列是否重复
            Set<Integer> queueSet = new HashSet<>(queues);
            if (queueSet.size() != queues.size()) {
                throw new QueueValidationException("Broker队列不能重复: " + entry.getKey());
            }
        }
    }

    /**
     * 检查Topic是否存在
     */
    private boolean topicExists(String topicName) {
        try {
            List<String> topicNames = rocketMQAdminClient.getAllTopicNames();
            return topicNames.contains(topicName);

        } catch (Exception e) {
            log.error("检查Topic是否存在失败: {}", topicName, e);
            return false;
        }
    }
}