用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();