MongoDB Stream是如何实现完美数据增量迁移的？

定义Channel频道的转换表：

public static enum Channel {  
    Food("美食"),  
    Emotion("情感"), 
    Pet("宠物"),  
    House("家居"),  
    Marriage("征婚"),  
    Education("教育"),  
    Travel("旅游") ; 
 
    private final String oldName;  
    public String getOldName() {  
        return oldName;  
    }  
    private Channel(String oldName) {  
        this.oldName = oldName;  
    }  
    /**  
     * 转换为新的名称  
     *  
     * @param oldName  
     * @return  
     */  
    public static String toNewName(String oldName) {  
        for (Channel channel : values()) {  
            if (channel.oldName.equalsIgnoreCase(oldName)) {  
                return channel.name();  
            }  
        }  
        return "";  
    }  
    /**  
     * 返回一个随机频道  
     *  
     * @return  
     */  
    public static Channel random() {  
        Channel[] channels = values();  
        int idx = (int) (Math.random() * channels.length);  
        return channels[idx];  
    }  
} 

为topic表预写入1w条记录：

开启监听任务，将topic上的所有变更写入到增量表：

代码中通过watch命令获得一个MongoCursor对象，用于遍历所有的变更。

FullDocument.UPDATE_LOOKUP选项启用后，在update变更事件中将携带完整的文档数据(FullDocument)。

watch()命令提交后，mongos会与分片上的mongod(主节点)建立订阅通道，这可能需要花费一点时间。

为了模拟线上业务的真实情况，启用几个线程对topic表进行持续写操作：

ChangeTask实现逻辑如下：

（编辑：晋中站长网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!