如何从ActiveMQ Classic移除已调度消息？Spring Boot函数问题排查

我想通过Spring Boot应用取消发送到ActiveMQ Classic队列的调度消息，自己写了移除消息的函数但没法正常工作。了解到调度消息时Broker不会返回可用于取消的ID，我试过从ActiveMQ控制台获取ID硬编码到函数里测试，但还是不行，麻烦帮忙排查scheduleMessageRemoval函数的问题：

@SpringBootApplication
@ComponentScan(basePackages = "com.example.paygokyu")
public class PaygokyuApplication {

    public static void main(String[] args) {
        ApplicationContext ctx = SpringApplication.run(PaygokyuApplication.class, args);

        JmsTemplate jms = ctx.getBean(JmsTemplate.class);

        // 发送延迟60秒的调度消息
        jms.convertAndSend("javainuse", "To be consumed with delay!!!!!!", message -> {
            String scheduledMessageId = UUID.randomUUID().toString();
            message.setLongProperty("AMQ_SCHEDULED_DELAY", 60000);
            message.setStringProperty("MY_SCHEDULED_MESSAGE_ID", scheduledMessageId);

            // 调用消息移除方法
            scheduleMessageRemoval(jms, message);

            return message;
        });
    }

    private static void scheduleMessageRemoval(JmsTemplate jmsTemplate, Message scheduledMessage) {
        try {
            // 获取JMS会话
            Session session = jmsTemplate.getConnectionFactory().createConnection()
                    .createSession(false, Session.AUTO_ACKNOWLEDGE);

            // 创建调度管理主题
            Destination management = session.createTopic(ScheduledMessage.AMQ_SCHEDULER_MANAGEMENT_DESTINATION);

            // 创建管理主题的消息生产者
            MessageProducer producer = session.createProducer(management);

            // 创建移除调度任务的消息
            Message remove = session.createMessage();

            // 设置移除操作的属性
            remove.setStringProperty(ScheduledMessage.AMQ_SCHEDULER_ACTION,
                    ScheduledMessage.AMQ_SCHEDULER_ACTION_REMOVE);

            String scheduledMessageID = scheduledMessage.getStringProperty("MY_SCHEDULED_MESSAGE_ID");

            System.out.println("Scheduled Message ID: " + scheduledMessageID);

            remove.setStringProperty(ScheduledMessage.AMQ_SCHEDULED_ID, scheduledMessageID);

            
            // 发送移除指令
            producer.send(remove);

            System.out.println("Scheduled Message REMOVED" );


            // 关闭会话
            session.close();
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

问题分析 #

1. 自定义ID无效：你用UUID.randomUUID()生成的MY_SCHEDULED_MESSAGE_ID不是ActiveMQ Broker分配的调度任务ID，Broker在接收调度消息后会生成专属的AMQ_SCHEDULED_ID，只有这个ID才能用于定位并删除任务，所以用自定义ID发送删除指令肯定找不到目标。
2. 资源泄漏：创建Connection后只关闭了Session，Connection没有关闭，会导致连接资源泄漏，长期运行会耗尽Broker连接池。
3. 同步时机问题：在发送调度消息的回调里立刻调用删除函数，此时Broker可能还没完成调度任务的创建流程，删除指令会因为找不到任务而直接失效。

修正方案 #

1. 获取正确的调度任务ID

发送调度消息后，登录ActiveMQ控制台（默认地址http://localhost:8161/admin/queues.jsp），找到目标队列javainuse，查看其下的调度消息列表，复制对应的Scheduled ID，这个才是需要传入删除函数的有效ID。

2. 修正删除函数

下面是调整后的代码，解决了资源泄漏和ID错误的问题，同时增加了延迟等待Broker创建任务的逻辑：

@SpringBootApplication
@ComponentScan(basePackages = "com.example.paygokyu")
public class PaygokyuApplication {

    public static void main(String[] args) {
        ApplicationContext ctx = SpringApplication.run(PaygokyuApplication.class, args);
        JmsTemplate jms = ctx.getBean(JmsTemplate.class);

        // 发送延迟60秒的调度消息
        jms.convertAndSend("javainuse", "To be consumed with delay!!!!!!", message -> {
            message.setLongProperty("AMQ_SCHEDULED_DELAY", 60000);
            return message;
        });

        // 延迟1秒再执行删除，确保Broker已完成调度任务创建
        try {
            Thread.sleep(1000);
            // 替换成控制台获取的真实调度ID
            String actualScheduledId = "控制台复制的Scheduled ID";
            scheduleMessageRemoval(jms, actualScheduledId);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    private static void scheduleMessageRemoval(JmsTemplate jmsTemplate, String scheduledTaskId) {
        Connection connection = null;
        Session session = null;
        MessageProducer producer = null;
        try {
            // 创建连接并启动
            connection = jmsTemplate.getConnectionFactory().createConnection();
            connection.start();
            // 创建会话
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            // 获取调度管理主题
            Destination managementTopic = session.createTopic(ScheduledMessage.AMQ_SCHEDULER_MANAGEMENT_DESTINATION);
            producer = session.createProducer(managementTopic);

            // 构建删除指令消息
            Message removeMessage = session.createMessage();
            removeMessage.setStringProperty(ScheduledMessage.AMQ_SCHEDULER_ACTION, ScheduledMessage.AMQ_SCHEDULER_ACTION_REMOVE);
            // 使用Broker分配的真实调度ID
            removeMessage.setStringProperty(ScheduledMessage.AMQ_SCHEDULED_ID, scheduledTaskId);

            // 发送删除指令
            producer.send(removeMessage);
            System.out.println("已发送删除指令，目标调度ID：" + scheduledTaskId);

        } catch (JMSException e) {
            System.err.println("删除调度消息失败：" + e.getMessage());
            e.printStackTrace();
        } finally {
            // 依次关闭所有JMS资源，避免泄漏
            try {
                if (producer != null) producer.close();
                if (session != null) session.close();
                if (connection != null) connection.close();
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    }
}

额外注意事项 #

• 确认ActiveMQ的调度功能已开启：默认是开启的，若修改过Broker配置，需确保broker.schedulerSupport=true。
• 删除指令必须在调度消息执行前发送，如果延迟时间已到、消息已被消费，删除操作无效。
• 生产环境中不要硬编码ID，可通过监听ActiveMQ的管理主题或使用JMX接口来动态获取调度任务ID。

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