Spring Boot 无法更新 azure cosmos db（MongoDb） 上的分片集合

Question

我在数据库中有一个集合“documentDev ”,其切分关键字为“dNumber”样本文档:

{
"_id" : "12831221wadaee23",
"dNumber" : "115",
"processed": false
}

如果我尝试使用任何查询工具更新这个文档，比如-

db.documentDev.update({
  "_id" : ObjectId("12831221wadaee23"),
  "dNumber":"115"
},{
    $set:{"processed": true}}, 
{ multi: false, upsert: false}
)}`

它会正确更新文档。但是如果我确实使用了Spring boot的mongorepository命令，如DocumentRepo.save（Object），它会引发异常

• 由以下原因引起：com.mongodb.MongoCommandException：命令失败，错误为61：“命令中的查询必须以单个分片键为目标”，在服务器 by3prdddc01-docdb-3.documents.azure.com:10255。完整的响应是 { “_t” ： “OKMongoResponse”， “ok” ： 0， “code” ： 61， “errmsg” ： “命令中的查询必须以单个分片键为目标”，“$err” ： “命令中的查询必须以单个分片键为目标” }

这是我的文档对象：

@Document(collection = "documentDev")
public class DocumentDev
{
@Id
private String id;
private String dNumber;
private String fileName;
private boolean processed;
}

这是我的知识库类-

@Repository
public interface DocumentRepo extends MongoRepository<DocumentDev, 
String> { }

以及我试图更新的值

< Li > Value:doc:{ " _ id ":" 12831221 Wada ee 23 "，" dNumber" : "115 "，" processed": true }

我尝试执行的函数：

@Autowired
DocumentRepo docRepo;

docRepo.save(doc); // Fails to execute

注意：我在dNumber字段上启用了分片。并且我能够成功地在非关系型数据库工具上使用本机查询进行更新。我还能够在非分片集合上执行存储库保存操作。

更新：我能够通过使用MongoTemplate创建本机查询来更新文档 - 我的查询看起来像这样 -

public DocumentDev updateProcessedFlag(DocumentDev request) {
    Query query = new Query();
    query.addCriteria(Criteria.where("_id").is(request.getId()));
    query.addCriteria(Criteria.where("dNumber").is(request.getDNumber()));
    Update update = new Update();
    update.set("processed", request.isProcessed());
    mongoTemplate.updateFirst(query, update, request.getClass());
    return request;
}

但这不是一个通用的解决方案，因为任何其他字段可能都有更新，我的文档也可能有其他字段。

Answer 1

我正在通过创建自定义存储库来破解它：

public interface CosmosCustomRepository<T> {
    void customSave(T entity);
    void customSave(T entity, String collectionName);
}

此存储库的实现：

public class CosmosCustomRepositoryImpl<T> implements CosmosCustomRepository<T> {

    @Autowired
    private MongoTemplate mongoTemplate;

    @Override
    public void customSave(T entity) {
        mongoTemplate.upsert(createQuery(entity), createUpdate(entity), entity.getClass());
    }

    @Override
    public void customSave(T entity, String collectionName) {
        mongoTemplate.upsert(createQuery(entity), createUpdate(entity), collectionName);
    }

    private Update createUpdate(T entity) {
        Update update = new Update();
        for (Field field : entity.getClass().getDeclaredFields()) {
            try {
                field.setAccessible(true);
                if (field.get(entity) != null) {
                    update.set(field.getName(), field.get(entity));
                }
            } catch (IllegalArgumentException | IllegalAccessException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
        return update;
    }

    private Query createQuery(T entity) {
        Criteria criteria = new Criteria();
        for (Field field : entity.getClass().getDeclaredFields()) {
            try {
                field.setAccessible(true);
                if (field.get(entity) != null) {
                    if (field.getName().equals("id")) {
                        return new Query(Criteria.where("id").is(field.get(entity)));
                    }
                    criteria.and(field.getName()).is(field.get(entity));
                }
            } catch (IllegalArgumentException | IllegalAccessException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
        return new Query(criteria);
    }
}

您的DocumentRepo将扩展此新的自定义存储库。

@Repository
public interface DocumentRepo extends MongoRepository<DocumentDev, String>, CosmosCustomRepository<DocumentDev> { }

要保存新文档，只需使用新建自定义保存

@Autowired
DocumentRepo docRepo;

docRepo.customSave(doc);

Answer 2

按照自定义存储库方法，我得到了一个错误，因为spring希望自定义实现｛EntityName｝CustomRepositoryImpl中有一个Cosmos实体，所以我重命名了该实现。我还添加了以下代码：

>

实体具有继承字段的情况

分片键并不总是Id，我们应该把它和id一起添加：{ “shardkeyName”： “shardValue” }

将生成的 ObjectId 添加到新文档的实体

 public class DocumentRepositoryImpl<T> implements CosmosRepositoryCustom<T> {

    @Autowired
    protected MongoTemplate mongoTemplate;

    @Override
    public T customSave(T entity) {
        WriteResult writeResult = mongoTemplate.upsert(createQuery(entity), createUpdate(entity), entity.getClass());
        setIdForEntity(entity,writeResult);
        return entity;
    }

    @Override
    public T customSave(T entity, String collectionName) {
        WriteResult writeResult = mongoTemplate.upsert(createQuery(entity), createUpdate(entity), collectionName);
        setIdForEntity(entity,writeResult);
        return entity;
    }

    @Override
    public void customSave(List<T> entities) {
        if(CollectionUtils.isNotEmpty(entities)){
            entities.forEach(entity -> customSave(entity));
        }
    }

    public <T> Update createUpdate(T entity){
        Update update = new Update();
        for (Field field : getAllFields(entity)) {
            try {
                field.setAccessible(true);
                if (field.get(entity) != null) {
                    update.set(field.getName(), field.get(entity));
                }
            } catch (IllegalArgumentException | IllegalAccessException e) {
                LOGGER.error("Error creating update for entity",e);
            }
        }
        return update;
    }

    public <T> Query createQuery(T entity) {
        Criteria criteria = new Criteria();
        for (Field field : getAllFields(entity)) {
            try {
                field.setAccessible(true);
                if (field.get(entity) != null) {
                    if (field.getName().equals("id")) {
                        Query query = new Query(Criteria.where("id").is(field.get(entity)));
                        query.addCriteria(new Criteria(SHARD_KEY_NAME).is(SHARD_KEY_VALUE));
                        return query;
                    }
                    criteria.and(field.getName()).is(field.get(entity));
                }
            } catch (IllegalArgumentException | IllegalAccessException e) {
                LOGGER.error("Error creating query for entity",e);
            }
        }
        return new Query(criteria);
    }

    private <T> List<Field> getAllFields(T entity) {
        List<Field> fields = new ArrayList<>();
        fields.addAll(Arrays.asList(entity.getClass().getDeclaredFields()));
        Class<?> c = entity.getClass().getSuperclass();
        if(!c.equals(Object.class)){
            fields.addAll(Arrays.asList(c.getDeclaredFields()));
        }
        return fields;
    }

    public <T> void setIdForEntity(T entity,WriteResult writeResult){
        if(null != writeResult && null != writeResult.getUpsertedId()){
            Object upsertId = writeResult.getUpsertedId();
            entity.setId(upsertId.toString());
        }
    }
}

我使用的是spring boot starter mongodb:1.5.1和spring数据mongodb:1.9.11

Answer 3

我有同样的问题，通过以下黑客解决：

@Configuration
public class ReactiveMongoConfig {

    @Bean
    public ReactiveMongoTemplate reactiveMongoTemplate(ReactiveMongoDatabaseFactory reactiveMongoDatabaseFactory,
            MongoConverter converter, MyService service) {
        return new ReactiveMongoTemplate(reactiveMongoDatabaseFactory, converter) {
            @Override
            protected Mono<UpdateResult> doUpdate(String collectionName, Query query, UpdateDefinition update,
                    Class<?> entityClass, boolean upsert, boolean multi) {
                query.addCriteria(new Criteria("shardKey").is(service.getShardKey()));
                return super.doUpdate(collectionName, query, update, entityClass, upsert, multi);
            }
        };
    }
}

最好有一个注释@ShardKey将文档字段标记为分片并将其自动添加到查询中。