Spring MongoDB中MongoChangeStream结合@SchedulerLock失效问题及替代方案咨询

一、为什么ShedLock的@SchedulerLock没生效？ #

核心原因是注解用错了场景：

• @SchedulerLock是用来拦截Spring @Scheduled标记的定时任务的，它的生效依赖Spring调度器触发方法执行。
• 你的insert方法是Mongo ChangeStream的回调方法，由MongoDB驱动触发，不是Spring定时任务，Spring不会扫描这个方法上的@Scheduled，自然不会触发ShedLock的锁逻辑，也就不会创建shedlock集合。

二、解决多实例重复执行ChangeStream事件的正确方案 #

ChangeStream本身会把事件推送给所有监听的实例，所以需要在处理每个事件前加分布式锁，确保同一事件仅被一个实例处理。

方案1：手动用ShedLock的LockManager加锁 #

放弃@SchedulerLock，直接在回调方法里手动获取锁，用事件的唯一标识（比如文档ID）作为锁名：

@Component
class BookChangeStreamListener(
    mongoTemplate: MongoTemplate,
    private val lockManager: LockManager
) : MongoChangeStreamListener<BookDocument>(mongoTemplate) {

    override fun createMessageListener() =
        createMessageListener(
            object : CallbackChangeStream<BookDocument> {
                override fun insert(raw: ChangeStreamDocument<Document>, body: BookDocument?) {
                    body?.let { book ->
                        // 用文档ID作为锁名，确保同一文档的事件只会被一个实例处理
                        val lockName = "BookChangeStream_Insert_${book.id}"
                        lockManager.lock(LockConfiguration(
                            name = lockName,
                            lockAtMostFor = Duration.ofMinutes(5),
                            lockAtLeastFor = Duration.ofMinutes(1)
                        )).ifPresent { lock ->
                            try {
                                // 执行你的业务逻辑
                            } finally {
                                // 必须释放锁
                                lock.unlock()
                            }
                        }
                    }
                }
            }
        )
}

方案2：用消息队列做事件分发（更优雅） #

把ChangeStream的事件转发到消息队列（如Kafka、RabbitMQ），利用队列的独占消费特性，天然避免多实例重复处理：

1. 用Spring Cloud Stream绑定Mongo ChangeStream作为数据源，将事件发送到队列
2. 多实例监听同一个队列，队列会自动把每个事件分发给一个实例处理

三、@SchedulerLock的替代方案 #

如果是针对定时任务的分布式锁需求，可选择以下方案：

• Quartz集群模式：自带分布式任务调度能力，基于数据库实现锁机制，支持任务分片
• Redisson分布式锁：基于Redis实现可重入锁，手动配合Spring定时任务使用
• Spring Integration分布式锁：支持Redis/ZooKeeper/Mongo等多种锁实现方式
• Elastic Job：轻量级分布式任务调度框架，支持分片、故障转移和锁机制

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