|
每一个变更任务会不断对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表
前者输出结果：
后者输出结果：
前后对比的结果是一致的。
五、后续优化 （编辑：核心网）
【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!
|
