Spring Boot中RabbitMQ监听器并发事件致重复对象的解决方案咨询

我在基于Spring Boot和Hibernate的应用中配置了RabbitMQ监听器，有时监听器会同时接收并处理两个事件，导致数据库中出现重复对象。我知道Inbox pattern是最优解决方案，但希望找到实现成本更低、耗时更短的替代方案。两个事件处理器操作同一张数据库表，我了解可以使用悲观锁，但这会降低系统性能。

我的AMQPListener代码如下：

@RabbitListener(
        queues = {RabbitApi.MY_EVENT_A + "." + RabbitApi.EXCHANGE},
        errorHandler = "amqpListenerExceptionHandler"
)
public void handleEvent_A(
        @Headers Map<String, Object> headers,
        @Payload @Valid EventA event
) {
    eEventHandlerA.handle(event);
}

@RabbitListener(
        queues = {RabbitApi.MY_EVENT_B + "." + RabbitApi.EXCHANGE},
        errorHandler = "amqpListenerExceptionHandler"
)
public void handleEvent_B(
        @Headers Map<String, Object> headers,
        @Payload @Valid EventB event
) {
    eEventHandlerB.handle(event);
}

补充说明：问题的根源在于两个事务重叠，第二个事件的处理器handleEvent_B在第一个事务提交前，无法感知到数据库中已创建的对象，进而重复创建了不应存在的对象。

1. 数据库唯一约束兜底 #

这是最直接的方案：在数据库表的业务唯一标识字段（或联合字段）上添加唯一约束。当第二个事务尝试插入重复对象时，数据库会抛出约束冲突异常，此时在事件处理器中捕获该异常，根据业务需求选择处理逻辑：

• 直接忽略，返回成功（因为目标对象已存在）
• 触发重试，等待第一个事务提交后再执行

在Hibernate中，可以通过实体类注解快速配置：

@Entity
@Table(uniqueConstraints = @UniqueConstraint(columnNames = {"biz_id", "type"}))
public class TargetEntity {
    // 业务唯一标识字段
    private String bizId;
    // 其他字段...
}

捕获异常时的处理逻辑示例：

try {
    targetRepository.save(entity);
} catch (SQLIntegrityConstraintViolationException e) {
    // 检查是否是唯一约束冲突，若是则跳过或重试
    log.info("对象已存在，跳过创建");
}

2. 乐观锁+重试机制 #

给实体类添加版本号字段（@Version），结合Spring Retry实现重试：

1. 处理器先根据业务标识查询数据库，若不存在则创建
2. 由于事务隔离级别限制，若第一个事务未提交，第二个查询会看不到数据，此时插入会触发重复，通过乐观锁异常或约束异常触发重试
3. 用Spring Retry给处理器添加重试逻辑，等待第一个事务提交后再执行

配置Spring Retry示例：

@Retryable(value = {SQLIntegrityConstraintViolationException.class}, maxAttempts = 3, backoff = @Backoff(delay = 500))
public void handleEvent_A(...) {
    eEventHandlerA.handle(event);
}

3. 单消费者限制（针对同业务消息） #

如果两个事件是针对同一业务实体的操作，可以限制对应队列的消费者并发数为1，确保同一时间只有一个线程处理该类消息：
在RabbitListener中指定并发数：

@RabbitListener(
        queues = {RabbitApi.MY_EVENT_A + "." + RabbitApi.EXCHANGE},
        errorHandler = "amqpListenerExceptionHandler",
        concurrency = "1"
)

注意：这种方式会降低整体消费吞吐量，仅适合业务并发不高的场景。

4. 轻量级分布式锁（Redis实现） #

用Redis实现分布式锁，基于业务唯一标识加锁，确保同一时间只有一个处理器能执行创建逻辑：

1. 处理事件前，用Redis的SETNX命令获取锁，设置合理的过期时间（略长于事务最大执行时间）
2. 获取锁成功则执行创建逻辑，失败则等待重试或直接返回
3. 事务完成后释放锁

用Spring Data Redis实现的简化示例：

@Autowired
private StringRedisTemplate redisTemplate;

public void handleEvent(Event event) {
    String lockKey = "lock:target:" + event.getBizId();
    Boolean lockAcquired = redisTemplate.opsForValue().setIfAbsent(lockKey, "locked", 30, TimeUnit.SECONDS);
    if (lockAcquired != null && lockAcquired) {
        try {
            // 查询并创建对象逻辑
            TargetEntity entity = targetRepository.findByBizId(event.getBizId());
            if (entity == null) {
                entity = new TargetEntity();
                // 设置属性
                targetRepository.save(entity);
            }
        } finally {
            redisTemplate.delete(lockKey);
        }
    } else {
        // 未获取到锁，触发重试或忽略
        log.info("锁被占用，稍后重试");
    }
}

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