如何使用ReactiveMongoOperation实现MongoDB自定义序列生成类

问题核心原因 #

你原有代码适配的是同步MongoOperations接口，而ReactiveMongoOperations的所有操作返回的都是响应式流类型（Mono/Flux），不会直接返回实体对象，所以同步取值的写法会报错。

前置准备：CustomSequences 实体类 #

先定义存储序列号的集合实体：

import org.springframework.data.annotation.Id;
import org.springframework.data.mongodb.core.mapping.Document;

@Document(collection = "custom_sequences")
public class CustomSequences {
    @Id
    private String id;
    private int seq;

    //  getter、setter 方法
    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }

    public int getSeq() {
        return seq;
    }

    public void setSeq(int seq) {
        this.seq = seq;
    }
}

修正后的 NextSequenceService 实现 #

标准响应式写法（推荐，适配全响应式调用链） #

直接返回Mono<Integer>，交给上游业务代码订阅处理，不要在服务层做阻塞：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.core.ReactiveMongoOperations;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;

import static org.springframework.data.mongodb.core.FindAndModifyOptions.options;

@Service
public class NextSequenceService {

    @Autowired
    private ReactiveMongoOperations reactiveMongoOperations;

    public Mono<Integer> getNextSequence(String seqName) {
        return reactiveMongoOperations.findAndModify(
                Query.query(Criteria.where("_id").is(seqName)),
                new Update().inc("seq", 1),
                options().returnNew(true).upsert(true),
                CustomSequences.class
            )
            .map(CustomSequences::getSeq)
            // 首次创建序列时默认返回1，可按需调整初始值
            .defaultIfEmpty(1);
    }
}

兼容阻塞取值写法（仅适配特殊场景，不推荐全响应式项目使用） #

如果你必须要拿到同步int返回值，可以用block()方法阻塞获取，注意不要在响应式线程池中调用该方法，会造成线程阻塞：

public int getNextSequenceBlock(String seqName) {
    return reactiveMongoOperations.findAndModify(
                Query.query(Criteria.where("_id").is(seqName)),
                new Update().inc("seq", 1),
                options().returnNew(true).upsert(true),
                CustomSequences.class
            )
            .map(CustomSequences::getSeq)
            .defaultIfEmpty(1)
            .block();
}

使用说明 #

• 首次调用对应seqName的方法时，会自动在custom_sequences集合插入对应记录，初始seq值为1，每次调用自增1
• 若需要自定义初始值，可以首次插入记录时手动指定seq值即可
• 该操作是原子性的，高并发场景下不会出现序列号重复问题

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