如何通过ActiveMQ Artemis管理API为HA集群所有Broker批量应用配置？

1. 监听连接故障转移，自动重执行配置 #

你提到ActiveMQConnectionFactory无法添加ExceptionListener，但实际上**ActiveMQConnection本身支持添加监听器**，可以在连接创建后注册ExceptionListener，当故障转移发生时触发配置重执行逻辑。

修改你的初始化代码，示例如下：

@Bean
public void initializeSettings() throws JMSException {
    QueueConnection connection = activeMQConnectionFactory.createQueueConnection();
    // 注册故障转移监听器
    connection.setExceptionListener(new ExceptionListener() {
        @Override
        public void onException(JMSException exception) {
            LOG.warn("连接故障，触发重连后重新执行配置", exception);
            // 重连后重新执行配置逻辑
            reapplyConfiguration();
        }
    });
    
    try (connection; QueueSession session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE)) {
        Queue managementQueue = session.createQueue("activemq.management");
        QueueRequestor requestor = new QueueRequestor(session, managementQueue);
        connection.start();
        
        // 初始执行配置
        applyConfiguration(session, requestor);
    }
}

// 抽离配置执行逻辑，方便复用
private void applyConfiguration(QueueSession session, QueueRequestor requestor) throws JMSException {
    Message addressMessage = session.createMessage();
    String json = """
            {
                "autoCreateQueues": false,
                "defaultMaxConsumers": 10
            }
            """;
    JMSManagementHelper.putOperationInvocation(addressMessage, ResourceNames.BROKER, "addAddressSettings", "address#", json);
    Message reply = requestor.request(addressMessage);
    
    if (JMSManagementHelper.hasOperationSucceeded(reply)) {
        LOG.info("Address settings applied successfully");
    } else {
        String errorMsg = JMSManagementHelper.getResult(reply, String.class);
        LOG.error("Failed to apply address settings: {}", errorMsg);
    }
}

// 重连后执行配置的方法
private void reapplyConfiguration() {
    try (QueueConnection connection = activeMQConnectionFactory.createQueueConnection();
         QueueSession session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE)) {
        Queue managementQueue = session.createQueue("activemq.management");
        QueueRequestor requestor = new QueueRequestor(session, managementQueue);
        connection.start();
        applyConfiguration(session, requestor);
    } catch (JMSException e) {
        LOG.error("Failed to reapply configuration after failover", e);
    }
}

故障转移后，新连接会指向集群中的另一个Broker（可能是新主节点），此时执行配置可确保该节点也应用了设置。

2. 利用Artemis配置同步特性 #

对于ActiveMQ Artemis 2.x的复制集群，你可以开启配置复制，让主节点的配置变更自动同步到从节点。

在broker.xml中配置：

<configuration>
    <core>
        <!-- 启用配置复制 -->
        <configuration-replication enabled="true"/>
        <!-- 其他核心配置 -->
        <cluster-connections>
            <cluster-connection name="my-cluster">
                <!-- 集群连接配置 -->
                <replication-cluster>true</replication-cluster>
            </cluster-connection>
        </cluster-connections>
    </core>
</configuration>

开启后，通过管理API在主节点执行的配置变更（如addAddressSettings）会自动同步到所有从节点，无需逐个执行，这是最省心的方案。

3. 基于集群发现批量执行配置 #

如果你不想依赖配置同步，可以通过Discovery Group获取集群中所有Broker的节点信息，然后逐个连接执行配置。

示例代码：

@Bean
public void initializeSettings() throws NamingException, JMSException {
    ArtemisTargetConfig config = (ArtemisTargetConfig) new InitialContext().lookup("java:comp/env/bean/ArtemisConfig");
    
    UDPBroadcastEndpointFactory udpCfg = new UDPBroadcastEndpointFactory();
    udpCfg.setGroupAddress(config.getGroupAddress()).setGroupPort(config.getGroupPort());
    DiscoveryGroupConfiguration groupConfiguration = new DiscoveryGroupConfiguration();
    groupConfiguration.setBroadcastEndpointFactory(udpCfg);
    
    // 获取集群中所有节点的连接器信息
    List<TransportConfiguration> brokerConnectors = groupConfiguration.getTransportConfigurations();
    
    for (TransportConfiguration connector : brokerConnectors) {
        // 为每个Broker创建单独的连接工厂
        ActiveMQConnectionFactory cf = ActiveMQJMSClient.createConnectionFactoryWithoutHA(connector, JMSFactoryType.CF);
        cf.setUser(config.getUserName());
        cf.setPassword(EncryptionUtils.decryptWebappConfig(config.getPassword()));
        
        // 执行配置
        try (QueueConnection connection = cf.createQueueConnection();
             QueueSession session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE)) {
            Queue managementQueue = session.createQueue("activemq.management");
            QueueRequestor requestor = new QueueRequestor(session, managementQueue);
            connection.start();
            
            applyConfiguration(session, requestor);
            LOG.info("Applied settings to broker: {}", connector.getName());
        } catch (JMSException e) {
            LOG.error("Failed to apply settings to broker: {}", connector.getName(), e);
        }
    }
}

// 复用之前的applyConfiguration方法
private void applyConfiguration(QueueSession session, QueueRequestor requestor) throws JMSException {
    // 配置执行逻辑...
}

这种方式会遍历集群中所有发现的Broker，确保每个节点都应用配置，但需要处理节点不可达的异常情况。

4. 借助Spring事件与定时任务兜底 #

如果以上方案仍有遗漏，可结合Spring的ApplicationReadyEvent和定时任务，定期检查并补全配置：

@Configuration
@EnableScheduling
public class JmsConfig {
    // ... 现有代码
    
    @EventListener(ApplicationReadyEvent.class)
    public void onApplicationReady() throws JMSException {
        // 应用启动时首次执行配置
        initializeSettings();
    }
    
    @Scheduled(fixedRate = 3600000) // 每小时执行一次
    public void periodicConfigurationCheck() {
        try {
            initializeSettings();
        } catch (JMSException e) {
            LOG.error("Periodic configuration check failed", e);
        }
    }
}

定时任务可以确保新加入集群的Broker也能被配置覆盖，适合动态扩展的集群场景。

内容的提问来源于stack exchange，提问作者Sergio Scaramuzzi