单消费者多用户场景下ActiveMQ消息留存与锁定策略咨询

针对你基于ActiveMQ 5.14和Camel 2.21的消息锁定+确认场景，我整理了几个适配你现有REST API架构的可行方案，都是结合你已经实现的逻辑来优化的：

这是对你初步思路的优化，利用ActiveMQ原生的延迟投递特性实现自动超时回滚：

• 拆分两个队列：machine-jobs（待处理主队列）和machine-jobs-locked（锁定队列）
• 当REST GET请求触发时，Camel路由从主队列独占消费（避免同一条消息被多个请求取走），给消息添加延迟投递属性后发送到锁定队列，同时把原消息ID返回给客户端
• 锁定队列的消息会在你设定的锁定时间后自动投递，此时Camel监听该队列并把消息转回主队列
• 收到REST DELETE请求时，根据客户端传入的消息ID，从锁定队列中找到对应消息并删除

关键配置与代码示例 #

1. 先在ActiveMQ的activemq.xml中开启调度支持：

<broker xmlns="http://activemq.apache.org/schema/core" brokerName="localhost" dataDirectory="${activemq.data}" schedulerSupport="true">
  <!-- 其他配置 -->
</broker>

2. Camel路由实现：

// GET请求获取消息并锁定
from("rest:get:/message")
    .to("activemq:machine-jobs?consumerType=Exclusive") // 独占消费确保单条消息仅被取一次
    .process(exchange -> {
        Message msg = exchange.getIn();
        // 设置5分钟锁定超时（单位：毫秒）
        msg.setHeader(ScheduledMessage.AMQ_SCHEDULED_DELAY, 5 * 60 * 1000);
        // 保留原消息ID，用于DELETE时匹配
        msg.setHeader("ORIGINAL_MSG_ID", msg.getMessageId());
    })
    .to("activemq:machine-jobs-locked")
    .transform().body(); // 返回消息内容给客户端

// 锁定超时后自动转回主队列
from("activemq:machine-jobs-locked")
    .to("activemq:machine-jobs");

// DELETE请求确认并删除消息
from("rest:delete:/message/{msgId}")
    .process(exchange -> {
        String targetId = exchange.getPathVariable("msgId");
        // 通过ActiveMQ API从锁定队列删除对应消息
        ActiveMQConnection conn = (ActiveMQConnection) exchange.getContext()
                .getRegistry().lookupByName("activeMQConnection");
        try (ActiveMQSession session = (ActiveMQSession) conn.createSession(false, Session.AUTO_ACKNOWLEDGE)) {
            ActiveMQQueue lockedQueue = (ActiveMQQueue) session.createQueue("machine-jobs-locked");
            session.browse(lockedQueue, message -> {
                if (targetId.equals(message.getHeader("ORIGINAL_MSG_ID"))) {
                    ((ActiveMQMessage) message).acknowledge();
                    return false; // 找到后停止遍历
                }
                return true;
            });
        }
    });

利用ActiveMQ事务的回滚特性，配合缓存实现锁定逻辑：

• REST GET接口对应的Camel路由使用事务性消费者，获取消息后不立即提交事务，而是把消息ID和锁定状态存入缓存（单机用内存，分布式建议用Redis）
• 设置缓存的过期时间等于锁定时长，同时启动定时任务监听缓存：如果收到DELETE请求，提交事务并删除缓存条目；如果缓存过期，回滚事务，消息自动回到主队列
• 注意：需要配置ActiveMQ的重发策略，避免超时回滚后消息频繁重复投递

核心要点 #

• 在Camel路由中开启事务：to("activemq:machine-jobs?transacted=true")
• 缓存中存储消息ID -> 锁定状态的键值对，过期时间设为锁定时长
• 事务回滚后，ActiveMQ会把消息重新放入队列，可通过maxRedeliveries和redeliveryDelay配置重发规则

用Camel的聚合组件实现“获取消息-等待确认”的配对逻辑：

• 以消息ID作为聚合关联ID，REST GET请求触发取消息后，将消息加入聚合组，设置超时时间（锁定时长）
• 当收到对应DELETE请求时，完成聚合并删除原消息；如果超时未收到确认，自动把消息放回主队列
• 分布式环境下，可将聚合存储配置为JDBC或Redis，避免单机内存聚合的集群问题

路由示例片段 #

from("rest:get:/message")
    .to("activemq:machine-jobs")
    .aggregate(header("JMSMessageID"), new AggregationStrategy() {
        @Override
        public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
            return newExchange; // 保留原始消息
        }
    })
    .completionTimeout(5 * 60 * 1000) // 5分钟锁定超时
    .completionPredicate(exchange -> exchange.getProperty("CONFIRMED", false)) // 收到DELETE标记为已确认
    .onCompletion().process(exchange -> {
        if (!exchange.getProperty("CONFIRMED", false)) {
            // 超时未确认，放回主队列
            exchange.getIn().setBody(exchange.getIn().getBody());
            exchange.getContext().createProducerTemplate().send("activemq:machine-jobs", exchange);
        }
    });

// DELETE请求标记聚合组为已确认
from("rest:delete:/message/{msgId}")
    .process(exchange -> {
        String msgId = exchange.getPathVariable("msgId");
        // 触发对应聚合组完成
        exchange.getContext().getExtension(AggregateComponent.class)
                .forceCompletionOfGroup(msgId);
        exchange.setProperty("CONFIRMED", true);
    });

• 无论哪种方案，都要确保用JMSMessageID作为消息的唯一标识，避免DELETE时匹配错误
• 分布式环境下，优先选择Redis等分布式缓存/存储，避免单机状态丢失导致的逻辑混乱
• ActiveMQ的独占消费者（consumerType=Exclusive）可以确保同一条消息不会被多个GET请求同时取走，是核心的并发控制手段

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