用spring mongo查询oplog时间戳

Question

我想查询使用Java的mongodbs oplog，如果可能的话用spring mongodb集成。我的问题是从Java创建以下查询：

db['oplog.rs'].find({ "ts": { $gt: Timestamp(1489568405,34) }, $and: [ { "ns": "myns" } ] }) 

我已经试过像BsonTimestamp或BSONTimestamp一些事情，导致错误的querys。使用

BasicQuery({ "ts": { $gt: Timestamp(1489568405,34) }, $and: [ { "ns": "myns" } ] }) 

导致java mongodb驱动程序的JSON解析器发生错误。

任何提示？

THX于尔根

一个典型的记录是这样的：

{ 
    "ts" : Timestamp(1489567144, 2), 
    "t" : NumberLong(2), 
    "h" : NumberLong(7303473893196954969), 
    "v" : NumberInt(2), 
    "op" : "i", 
    "ns" : "asda.jam", 
    "o" : { 
     "_id" : NumberInt(2), 
     "time" : ISODate("2017-03-15T08:39:00.000+0000"), 
     "roadDesc" : { 
      "roadId" : NumberInt(28102917), 
      "roadName" : "A480 W" 
     }, 
     "posUpFront" : NumberInt(1003), 
     "posDownFront" : NumberInt(1003), 
     "_class" : "de.heuboe.acaJNI.test.Jam" 
    } 
} 

Answer 1

你可以尝试像下面。

Query query = Query.query(Criteria.where("ts").gt(new BSONTimestamp(1489568405, 34)).andOperator(Criteria.where("ns").is("myns"))); 
List<BasicDBObject> basicDBObjects = mongoTemplate.find(query, BasicDBObject.class, "oplog.rs"); 

Answer 2

对于类似于NumberLong，Timestamp等的构造，Mongo具有扩展的JSON语法，它在Mongo shell上工作。为了使它在Java代码中工作，它们具有严格的JSON模式，其中这些运算符使用JSON表示（https://docs.mongodb.com/manual/reference/mongodb-extended-json/#bson-data-types-and-associated-representations）。要使用Java来执行此操作，您可以创建一个自定义转换器并将其注册到MappingMongoConverter中（请参阅下面的代码片段）。转换器应该将数据类型（比如BSONTimestamp）转换为适当的严格JSON文档格式。

@WritingConverter 
public class BsonTimestampToDocumentConverter implements Converter<BSONTimestamp, Document> { 

    private static final Logger LOGGER = LoggerFactory.getLogger(BsonTimestampToDocumentConverter.class); 

    public BsonTimestampToDocumentConverter() { 
    // 
    } 

    @Override 
    public Document convert(BSONTimestamp source) { 
    LOGGER.trace(">>>> Converting BSONTimestamp to Document"); 
    Document value = new Document(); 
    value.put("t", source.getTime()); 
    value.put("i", source.getInc()); 
    return new Document("$timestamp", value); 
    } 
} 

注册它在MappingMongoConverter这样

public MappingMongoConverter syncLocalMappingMongoConverter() throws Exception { 
    MongoMappingContext mappingContext = new MongoMappingContext(); 
    DbRefResolver dbRefResolver = new DefaultDbRefResolver(syncLocalDbFactory()); 
    MappingMongoConverter converter = new MappingMongoConverter(dbRefResolver, mappingContext); 
    converter.setCustomConversions(customConversions()); 

    return converter; 
} 


    private CustomConversions customConversions() { 
    List<Converter<?, ?>> converterList = new ArrayList<>(); 
    converterList.add(new BsonTimestampToDocumentConverter()); 
    // add the other converters here 
    return new CustomConversions(CustomConversions.StoreConversions.NONE, converterList); 
} 

此处，我用它来查询OPLOG库一定时间后返回记录的例子（资料库同步被用来区分它来自我正在处理的反应性异步内容。异步存储库看起来完全一样，除了它应该扩展ReactiveMongoRepository）。 OplogRecord类是我创建的Java bean，用于匹配MongoDb oplog记录的结构。

public interface SyncOplogRepository extends MongoRepository<OplogRecord, Long> { 

    @Query(value = "{ \"op\": { $nin: ['n', 'c'] } }") List<OplogRecord> findRecordsNotEqualToNOrC(); 

    @Query(value = "{'ts' : {$gte : ?0}, \"op\": { $nin: ['n', 'c'] } }") 
    List<OplogRecord> findRecordsNotEqualToNOrCAfterTime(BSONTimestamp timestamp); 

    @Query(value = "{'ts' : {$lt : ?0}, \"op\": { $nin: ['n', 'c'] } }") 
    List<OplogRecord> findRecordsNotEqualToNOrCBeforeTime(BSONTimestamp timestamp); 

} 

OplogRecord类

import com.mongodb.DBObject; 
import org.bson.BsonTimestamp; 
import org.bson.types.BSONTimestamp; 
import org.slf4j.Logger; 
import org.slf4j.LoggerFactory; 
import org.springframework.data.annotation.Id; 
import org.springframework.data.mongodb.core.mapping.Document; 

import java.util.Map; 


@Document(collection = "oplog.rs") 
public class OplogRecord { 

    @Id 
    private Long id; 

    /** 
    * Timestamp 
    */ 
    private BsonTimestamp ts; 

    /** 
    * Unique id for this entry 
    */ 
    private Long h; 

    /** 
    * DB and collection name of change. 
    */ 
    private String ns; 

    /** 
    * The actual document that was modified/inserted/deleted 
    */ 
    private Map<String, Object> o; 

    /** 
    * The operation that was performed 
    */ 
    private String op; 

    /** 
    * ?? 
    */ 
    private Long t; 

    /** 
    * ?? 
    */ 
    private Integer v; 

    public BsonTimestamp getTs() { 
    return ts; 
    } 

    public void setTs(BsonTimestamp ts) { 
    this.ts = ts; 
    } 

    public Long getH() { 
    return h; 
    } 

    public void setH(Long h) { 
    this.h = h; 
    } 

    public String getNs() { 
    return ns; 
    } 

    public void setNs(String ns) { 
    this.ns = ns; 
    } 

    public Map<String, Object> getO() { 
    return o; 
    } 

    public void setO(Map<String, Object> o) { 
    this.o = o; 
    } 

    public String getOp() { 
    return op; 
    } 

    public void setOp(String op) { 
    this.op = op; 
    } 

    public Long getT() { 
    return t; 
    } 

    public void setT(Long t) { 
    this.t = t; 
    } 

    public Integer getV() { 
    return v; 
    } 

    public void setV(Integer v) { 
    this.v = v; 
    } 
} 

〜
〜

Answer 3

可以过滤使用org.bson.BsonTimestamp。

BsonTimestamp lastReadTimestamp = new BsonTimestamp(1489568405, 34); 
Bson filter = new Document("$gt", lastReadTimestamp); 

然后您可以使用发现，像这样，

oplogColl.find(new Document("ts", filter)); 

或者你可以创建一个保释的游标，并通过这样的文件迭代，

MongoCursor oplogCursor = 
        oplogColl 
          .find(new Document("ts", filter)) 
          .cursorType(CursorType.TailableAwait) 
          .noCursorTimeout(true) 
          .batchSize(1000) 
          .iterator();