|
每一个改观使命会不绝对topic发生写操纵，触发一系列ChangeEvent发生：
- doInsert：天生随机频道的topic后，执行insert；
- doUpdate：随机取得一个topic，将其channel字段改为随机值，执行update；
- doReplace：随机取得一个topic，将其channel字段改为随机值，执行replace；
- doDelete：随机取得一个topic，执行delete。
以doUpdate为例，实当代码如下：
启动一个全量迁徙使命，将topic表中数据迁徙到topic_new新表：
在全量迁徙开始前，先获适合前时候的的最大 _id 值（可以将此值记录下来）作为终点，随后逐个完成迁徙转换。
在全量迁徙完成后，便开始最后一步：增量迁徙。
注：增量迁徙进程中，改观操纵如故在举办。
- final MongoCollection<Document> topicIncrCollection = getCollection(coll_topic_incr);
- final MongoCollection<Document> topicNewCollection = getCollection(coll_topic_new);
- ObjectId currentId = null;
- Document sort = new Document("_id", 1);
- MongoCursor<Document> cursor = null;
- // 批量巨细
- int batchSize = 100;AtomicInteger count = new AtomicInteger(0);
- try {
- while (true) {
- boolean isWatchTaskStillRunning = watchFlag.getCount() > 0;
- // 按ID增量分段拉取
- if (currentId == null) {
- cursor = topicIncrCollection.find().sort(sort).limit(batchSize).iterator();
- } else {
- cursor = topicIncrCollection.find(new Document("_id", new Document("$gt", currentId)))
- .sort(sort).limit(batchSize).iterator();
- }
- boolean hasIncrRecord = false;
- while (cursor.hasNext()) {
- hasIncrRecord = true;
- Document incrDoc = cursor.next();
- OperationType opType = OperationType.fromString(incrDoc.getString(field_op));
- ObjectId docId = incrDoc.getObjectId(field_key);
- // 记录当前ID
- currentId = incrDoc.getObjectId("_id");
- if (opType == OperationType.DELETE) {
- topicNewCollection.deleteOne(new Document("_id", docId));
- } else {
- Document doc = incrDoc.get(field_data, Document.class);
- // channel转换
- String oldChannel = doc.getString(field_channel);
- doc.put(field_channel, Channel.toNewName(oldChannel));
- // 启用upsert
- UpdateOptions options = new UpdateOptions().upsert(true);
- topicNewCollection.replaceOne(new Document("_id", docId),
- incrDoc.get(field_data, Document.class), options);
- }
- if (count.incrementAndGet() % 10 == 0) {
- logger.info("IncrTransferTask progress, count: {}", count.get());
- }
- }
- // 当watch遏制事变(没有更多改观)，同时也没有必要处理赏罚的记录时，跳出
- if (!isWatchTaskStillRunning && !hasIncrRecord) {
- break;
- }
- sleep(200);
- }
- } catch (Exception e) {
- logger.error("IncrTransferTask ERROR", e);
- }
增量迁徙的实现是一个不绝tail的进程，操作 **_id 字段的有序特征 ** 举办分段迁徙；即记录下当前处理赏罚的_id值，轮回拉取在该_id值之后的记录举办处理赏罚。
增量表(topic_incr)中除了DELETE改观之外，别的的范例都保存了整个文档，因此可直接操作replace + upsert追加到新表。
最后，运行整个措施。
查察topic表和topic_new表，发明两者数目是沟通的。为了进一步确认同等性，我们对两个表的别离做一次聚合统计：
topic表
topic_new表
前者输出功效：
后者输出功效：
前后比拟的功效是同等的。
五、后续优化 （编辑：湖南网）
【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!
|
