多实例Spring Boot应用中Oracle AQ同组消息顺序处理方案问询

针对你的场景——Spring Boot集成Oracle AQ，要求同组消息严格顺序处理、不同组消息并行，单实例下通过选择器实现但多实例失效的问题，以下是几个可行的解决方案：

方案一：利用Oracle AQ原生JMSXGroupID特性 #

Oracle AQ支持标准JMS的JMSXGroupID属性，该特性会将同一分组的消息绑定到同一个消费者会话，确保同组消息只会被一个线程/实例处理，天然保证顺序；不同组消息则可分配到不同实例/线程实现并行。

修改代码： #

1. 生产者设置JMSXGroupID

替换自定义的requestDtoGroup属性为标准JMSXGroupID：

@Component
public class MessageProducer {

    private final JmsTemplate jmsTemplate;

    public MessageProducer(JmsTemplate jmsTemplate) {
        this.jmsTemplate = jmsTemplate;
    }

    public void sendMessage(RequestDto requestDto) {
        jmsTemplate.convertAndSend(
                "test_queue",
                requestDto,
                message -> {
                    // 改用标准JMS分组属性
                    message.setStringProperty("JMSXGroupID", requestDto.getGroup());
                    return message;
                }
        );
    }
}

2. 统一消费者监听

无需为每个分组单独写监听方法，一个通用监听即可：

@Component
@Log4j2
public class MessageConsumer {

    @JmsListener(destination = "test_queue")
    public void listen(Message message) throws JMSException {
        String groupId = message.getStringProperty("JMSXGroupID");
        log.info("Processing message from group: {}", groupId);
        // 业务处理逻辑，确保事务提交/回滚正确
    }
}

3. 调整JMS容器工厂配置

设置合理的并发度，同时确保事务和超时配置：

@Configuration
@EnableJms
public class JmsConfiguration {

    @Bean
    public ConnectionFactory connectionFactory() throws JMSException {
        Properties properties = new Properties();
        properties.setProperty("user", "AQ_USER");
        properties.setProperty("password", "your_password");

        return AQjmsFactory.getQueueConnectionFactory(
                "jdbc:oracle:thin:@localhost:1521/ORCLPDB1",
                properties
        );
    }

    @Bean
    public DefaultJmsListenerContainerFactory jmsListenerContainerFactory() throws JMSException {
        var containerFactory = new DefaultJmsListenerContainerFactory();
        containerFactory.setConnectionFactory(connectionFactory());
        containerFactory.setSessionAcknowledgeMode(Session.SESSION_TRANSACTED);
        containerFactory.setSessionTransacted(true);
        // 设置并发线程数，根据预期分组数量调整（如1-5表示最少1个，最多5个并发线程）
        containerFactory.setConcurrency("1-5");
        // 设置消息接收超时，避免线程长期阻塞
        containerFactory.setReceiveTimeout(Duration.ofSeconds(5).toMillis());
        return containerFactory;
    }
}

原理：Oracle AQ会根据JMSXGroupID将同组消息路由到同一个消费者会话，因此同组消息只会被一个线程处理；多实例环境下，不同组的消息会被分配到不同实例的线程，实现并行处理。

方案二：分布式锁+消息选择器（备选方案） #

如果Oracle AQ的JMSXGroupID特性因版本或环境限制无法生效，可通过分布式锁控制同组消息的消费逻辑，确保同一时间只有一个实例处理某组消息。

示例代码（使用Redisson实现分布式锁）： #

1. 引入Redisson依赖（pom.xml）

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson-spring-boot-starter</artifactId>
    <version>3.23.3</version>
</dependency>

2. 修改消费者逻辑

@Component
@Log4j2
public class MessageConsumer {

    private final RedissonClient redissonClient;

    public MessageConsumer(RedissonClient redissonClient) {
        this.redissonClient = redissonClient;
    }

    @JmsListener(destination = "test_queue")
    public void listen(Message message) throws JMSException, InterruptedException {
        String groupId = message.getStringProperty("requestDtoGroup");
        // 针对每个分组创建唯一锁
        RLock lock = redissonClient.getLock("aq_group_lock:" + groupId);
        
        try {
            // 尝试获取锁：等待10秒，持有锁30秒（根据业务处理时长调整）
            if (lock.tryLock(10, 30, TimeUnit.SECONDS)) {
                log.info("Processing message from group: {}", groupId);
                // 业务处理逻辑
            } else {
                // 获取锁失败，抛出异常触发消息回滚，重新入队等待处理
                throw new RuntimeException("Failed to acquire lock for group: " + groupId);
            }
        } finally {
            // 确保锁被当前线程持有才释放
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }
}

注意：需确保消息处理失败时事务回滚，让消息重新入队；锁的超时时间需大于业务处理的最大耗时，避免锁提前释放导致并发处理。

方案三：队列分区（固定分组场景） #

如果业务分组是固定的，可创建多个Oracle AQ队列，每个分组对应一个队列。生产者根据groupId将消息发送到对应队列，每个队列配置单线程监听器，多实例下通过JMS独占消费者特性确保每个队列只有一个实例消费。

该方案适合分组固定的场景，缺点是分组动态变化时需要动态创建队列和监听器，维护成本较高。

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