基于Mongodb分布式锁解决定时任务并发执行问题

目录

• 前言
• All in one Code
• 原理

@Component
@Slf4j
public class MongoDBLock {

    private static final int DEFAULT_LOCK_TIMEOUT = 30;//锁的默认超时时间，单位秒

    private MongoTemplate mongoTemplate;
    private int lockTimeout;

    public MongoDBLock(MongoTemplate mongoTemplate) {
        this.mongoTemplate = mongoTemplate;
        this.lockTimeout = DEFAULT_LOCK_TIMEOUT;
    }

    /**
     * 尝试获取分布式锁
     *
     * @param lockKey 锁的key
     * @return true：获取锁成功，false：获取锁失败
     */
    private boolean acquireLock(String lockKey) {
        LockDocument document = new LockDocument();
        document.setId(lockKey);
        document.setExpireAt(Instant.ofEpochMilli(Instant.now().toEpochMilli() + lockTimeout * 1000));
        try {
            mongoTemplate.insert(document);
            return true;
        } catch (Exception e) {

        }
        return false;
    }

    /**
     * 释放分布式锁
     *
     * @param lockKey 锁的key
     */
    private void releaseLock(String lockKey) {
        Query query = new Query(Criteria.where("key").is(lockKey));
        mongoTemplate.remove(query, LockDocument.class);
        log.info("程序执行成功，释放分布式锁，lockKey:{}",lockKey);
    }

    /**
     * 分布式锁入口方法，参数lockName为锁的名称，lockKey为需要加锁的key，执行完成后自动释放锁
     *
     * @param lockKey
     * @param task
     * @param <T>
     * @throws Exception
     */
    public <T> void executeWithLock(String lockKey, ITask<T> task) throws Exception {
        boolean locked = acquireLock(lockKey);
        if (locked) {
            log.info("获取分布式锁成功，lockKey:{}",lockKey);
            try {
                task.execute();
            } finally {
                releaseLock(lockKey);
            }
        } else {
            log.warn("获取分布式锁失败，lockKey:{}", lockKey);
            throw new AppException("获取分布式锁失败！");
        }
    }

    @Data
    @Document(collection = "lock_collection")
    static class LockDocument {
        @Id
        private String id;
        @Indexed(expireAfterSeconds = DEFAULT_LOCK_TIMEOUT)
        private Instant expireAt;
    }

    @FunctionalInterface
    public interface ITask<T> {
        T execute() throws Exception;
    }
}

    @Resource
    MongoDBLock mongoDBLock;

    mongoDBLock.executeWithLock("key", () -> {
        // do some thing
        return null;
    });