如何通过Core协议远程管理Apache Artemis 2.35.0服务器？

核心逻辑概述 #

Core协议下的管理操作无需HTTP，直接通过Artemis原生Core客户端API与服务器交互，核心是借助ClientSession结合内置管理工具类，获取QueueControl等管理对象并执行操作。

1. 建立Core协议连接 #

首先创建带管理权限的Core连接，需使用具备manage权限的账号（默认admin/admin，需与服务器配置匹配）：

import org.apache.activemq.artemis.api.core.client.*;

// 初始化Core协议定位器，指定服务器地址
ServerLocator locator = ActiveMQClient.createServerLocator("tcp://your-artemis-host:61616");
locator.setUser("admin");
locator.setPassword("admin");

// 创建会话工厂与自动提交会话（管理操作推荐用事务会话，此处简化示例）
ClientSessionFactory factory = locator.createSessionFactory();
ClientSession session = factory.createSession(true, true);

2. 获取所有队列的QueueControl对象 #

先查询所有队列名称，再为每个队列生成对应的QueueControl代理实例：

import org.apache.activemq.artemis.api.core.management.*;

// 构造查询所有队列的管理请求
ClientMessage request = session.createMessage(false);
ManagementHelper.putOperationInvocation(request, "broker", "getQueueNames", "any");

// 发送请求到管理地址并接收回复
ClientProducer producer = session.createProducer(ManagementHelper.MANAGEMENT_ADDRESS);
ClientConsumer consumer = session.createConsumer(ManagementHelper.MANAGEMENT_RESPONSE_ADDRESS);
producer.send(request);
ClientMessage response = consumer.receive(5000);

// 解析回复得到队列名称数组
String[] queueNames = (String[]) ManagementHelper.getResult(response);

// 批量生成QueueControl实例
for (String queueName : queueNames) {
    QueueControl queueControl = ManagementHelper.createQueueControl(session, queueName);
    // 示例：获取队列消息数量
    System.out.printf("队列 %s 消息数: %d%n", queueName, queueControl.getMessageCount());
}

3. 在队列间移动消息 #

直接调用QueueControl的moveMessages方法即可实现消息迁移，也可通过管理命令直接发送请求：

// 方式1：通过QueueControl实例操作
QueueControl sourceQueue = ManagementHelper.createQueueControl(session, "source-queue");
// 参数：目标队列名、消息选择器（null表示所有消息）
sourceQueue.moveMessages("target-queue", null);

// 方式2：直接发送管理命令（适合批量或自定义场景）
ClientMessage moveRequest = session.createMessage(false);
ManagementHelper.putOperationInvocation(moveRequest, "queue", "source-queue", "moveMessages", "target-queue", null);
producer.send(moveRequest);
ClientMessage moveResponse = consumer.receive(5000);
// 验证操作结果
boolean isSuccess = (Boolean) ManagementHelper.getResult(moveResponse);

注意事项 #

• 确保服务器的TCP acceptor（默认61616端口）开启，且账号拥有manage权限
• 操作完成后需关闭session、factory、locator释放资源
• 复杂迁移操作建议使用事务会话，保证操作原子性

内容的提问来源于stack exchange，提问作者Geert Schuring