【Mongdb之数据同步篇】什么是Oplog、Mongodb 开启oplog，java监听oplog并写入关系型数据库、Mongodb动态切换数据源

Oplog是什么

oplog简介

oplog是local库下的一个固定集合，Secondary就是通过查看Primary 的oplog这个集合来进行复制的。每个节点都有oplog，记录这从主节点复制过来的信息，这样每个成员都可以作为同步源给其他节点。

Oplog 可以说是Mongodb Replication的纽带了。

副本集数据同步的过程

副本集中数据同步的详细过程：Primary节点写入数据，Secondary通过读取Primary的oplog得到复制信息，开始复制数据并且将复制信息写入到自己的oplog。如果某个操作失败（只有当同步源的数据损坏或者数据与主节点不一致时才可能发生），则备份节点停止从当前数据源复制数据。如果某个备份节点由于某些原因挂掉了，当重新启动后，就会自动从oplog的最后一个操作开始同步，同步完成后，将信息写入自己的oplog，由于复制操作是先复制数据，复制完成后再写入oplog，有可能相同的操作会同步两份，不过MongoDB在设计之初就考虑到这个问题，将oplog的同一个操作执行多次，与执行一次的效果是一样的。

作用：
　　当Primary进行写操作的时候，会将这些写操作记录写入Primary的Oplog 中，而后Secondary会将Oplog 复制到本机并应用这些操作，从而实现Replication的功能。
　　同时由于其记录了Primary上的写操作，故还能将其用作数据恢复。
　　可以简单的将其视作Mysql中的binlog。

Oplog的增长速度

oplog是固定大小，他只能保存特定数量的操作日志，通常oplog使用空间的增长速度跟系统处理写请求的速度相当，如果主节点上每分钟处理1KB的写入数据，那么oplog每分钟大约也写入1KB数据。如果单次操作影响到了多个文档（比如删除了多个文档或者更新了多个文档）则oplog可能就会有多条操作日志。db.testcoll.remove() 删除了1000000个文档，那么oplog中就会有1000000条操作日志。如果存在大批量的操作，oplog有可能很快就会被写满了。

大小：
　　Oplog 是一个capped collection。
　　在64位的Linux, Solaris, FreeBSD, and Windows 系统中，Mongodb默认将其大小设置为可用disk空间的5%（默认最小为1G，最大为50G），或也可以在mongodb复制集实例初始化之前将mongo.conf中oplogSize设置为我们需要的值。

local.oplog.rs 一个capped collection集合.可在命令行下使用–oplogSize 选项设置该集合大小尺寸.
　　但是由于Oplog 其保证了复制的正常进行，以及数据的安全性和容灾能力。

Oplog注意事项：

local.oplog.rs特殊的集合。用来记录Primary节点的操作。

为了提高复制的效率，复制集中的所有节点之间会相互的心跳检测（ping）。每个节点都可以从其他节点上获取oplog。

oplog中的一条操作。不管执行多少次效果是一样的

oplog的大小

第一次启动复制集中的节点时，MongoDB会建立Oplog,会有一个默认的大小，这个大小取决于机器的操作系统

rs.printReplicationInfo() 查看 oplog 的状态，输出信息包括 oplog 日志大小，操作日志记录的起始时间。

db.getReplicationInfo() 可以用来查看oplog的状态、大小、存储的时间范围。

开启Oplog

windows mongodb bin目录下找到配置文件/bin/mongod.cfg,配置如下：

replication:
  replSetName: local
  oplogSizeMB: 1024

双击mongo.exe


执行

rs.initiate({_id: "local", members: [{_id: 0, host: "localhost:27017"}]})

若出现如下情况则成功

{
	"ok" : 1,
	"operationTime" : Timestamp(1627503341, 1),
	"$clusterTime" : {
		"clusterTime" : Timestamp(1627503341, 1),
		"signature" : {
			"hash" : BinData(0,"AAAAAAAAAAAAAAAAAAAAAAAAAAA="),
			"keyId" : NumberLong(0)
		}
	}
}

监听Oplog日志

pom

 	<parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.7.10</version>
        <relativePath/>
    </parent>
    
		<dependency>
        	<groupId>org.springframework.boot</groupId>
       	 	<artifactId>spring-boot-starter-data-mongodb</artifactId>
   	 	</dependency>
 		 <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-jpa</artifactId>
        </dependency>
        <dependency>
            <groupId>org.mongodb</groupId>
            <artifactId>mongodb-driver</artifactId>
            <version>3.12.7</version>
        </dependency>
        <dependency>
            <groupId>com.vividsolutions</groupId>
            <artifactId>jts</artifactId>
            <version>1.13</version>
        </dependency>
        <dependency>
            <groupId>org.hibernate</groupId>
            <artifactId>hibernate-spatial</artifactId>
            <version>5.3.0.Beta1</version>
        </dependency>
        <dependency>
            <groupId>org.hibernate</groupId>
            <artifactId>hibernate-java8</artifactId>
            <version>5.3.0.Beta1</version>
        </dependency>
        <dependency>
            <groupId>com.bedatadriven</groupId>
            <artifactId>jackson-datatype-jts</artifactId>
            <version>2.3</version>
        </dependency>
        <dependency>
            <groupId>org.postgresql</groupId>
            <artifactId>postgresql</artifactId>
            <scope>runtime</scope>
        </dependency>

配置

spring.datasource.driver-class-name=org.postgresql.Driver
spring.datasource.url=jdbc:postgresql://localhost:5432/databaseName?characterEncoding=UTF-8&useUnicode=true&useSSL=false&tinyInt1isBit=false&currentSchema=public
spring.datasource.username=postgres
spring.datasource.password=123456
spring.jpa.database=postgresql
spring.jpa.show-sql=true
spring.jpa.hibernate.ddl-auto=update
spring.jpa.properties.hibernate.dialect=org.hibernate.spatial.dialect.postgis.PostgisDialect
server.port=10050
spring.data.mongodb.uri=mongodb://admin:123456@localhost:27017/?authSource=admin
spring.data.mongodb.database=databseName

代码

  import com.mongodb.CursorType;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoCursor;
import com.mongodb.client.MongoDatabase;
import com.mongodb.client.model.Filters;
import com.mongodb.util.JSON;
import lombok.extern.slf4j.Slf4j;
import org.bson.BsonTimestamp;
import org.bson.Document;
import org.bson.conversions.Bson;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import javax.persistence.EntityManager;
import javax.persistence.Query;

@Slf4j
@Component
public class OplogListener implements ApplicationListener<ContextRefreshedEvent> {
    @Resource
    private MongoTemplate mongoTemplate;
    @Resource
    private EntityManager entityManager;


    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        MongoDatabase db = mongoTemplate.getMongoDatabaseFactory().getMongoDatabase("local");
        MongoCollection<Document> oplog = db.getCollection("oplog.rs");

        BsonTimestamp startTS = getStartTimestamp();
        BsonTimestamp endTS = getEndTimestamp();

        Bson filter = Filters.and(Filters.gt("ts", startTS));

        MongoCursor<Document> cursor = oplog.find(filter).cursorType(CursorType.TailableAwait).iterator();

        while (true) {
            if (cursor.hasNext()) {
                Document doc = cursor.next();
                String operation = doc.getString("op");

                if (!"n".equals(operation)) {
                    String namespace = doc.getString("ns");
                    String[] nsParts = StringUtils.split(namespace, ".");
                    String collectionName = nsParts[1];
                    String databaseName = nsParts[0];
                    Document object = (Document) doc.get("o");
                    log.info("同步数据：databse-{}  collention-{}  data-{}", databaseName, collectionName, object);
                    if ("i".equals(operation)) {
                        insert((Document) doc.get("o"), databaseName, collectionName);
                    } else if ("u".equals(operation)) {
                        update((Document) doc.get("o"), (Document) doc.get("o2"), databaseName, collectionName);
                    } else if ("d".equals(operation)) {
                        delete((Document) doc.get("o"), databaseName, collectionName);
                    }
                }
            }
        }
    }

    private BsonTimestamp getStartTimestamp() {
        long currentSeconds = System.currentTimeMillis() / 1000;
        return new BsonTimestamp((int) currentSeconds, 1);
    }

    private BsonTimestamp getEndTimestamp() {
        return new BsonTimestamp(0, 1);
    }

    private void insert(Document object, String databaseName, String collectionName) {
        entityManager.getTransaction().begin();
        try {
            String json = JSON.serialize(object);
            Query query = entityManager.createNativeQuery("INSERT INTO " + collectionName + " (json) VALUES (:json)");
            query.setParameter("json", json);
            query.executeUpdate();
            entityManager.getTransaction().commit();
        } catch (Exception e) {
            entityManager.getTransaction().rollback();
            throw new RuntimeException(e);
        }
    }

    private void update(Document object, Document update, String databaseName, String collectionName) {
        entityManager.getTransaction().begin();
        try {
            String json = JSON.serialize(object);
            String updateJson = JSON.serialize(update);
            Query query = entityManager.createNativeQuery("UPDATE " + collectionName + " SET json = :json WHERE json = :updateJson");
            query.setParameter("json", json);
            query.setParameter("updateJson", updateJson);
            query.executeUpdate();
            entityManager.getTransaction().commit();
        } catch (Exception e) {
            entityManager.getTransaction().rollback();
            throw new RuntimeException(e);
        }
    }

    private void delete(Document object, String databaseName, String collectionName) {
        entityManager.getTransaction().begin();
        try {
            String json = JSON.serialize(object);
            Query query = entityManager.createNativeQuery("DELETE FROM " + collectionName + " WHERE json = :json");
            query.setParameter("json", json);
            query.executeUpdate();
            entityManager.getTransaction().commit();
        } catch (Exception e) {
            entityManager.getTransaction().rollback();
            throw new RuntimeException(e);
        }
    }
}

动态切换Mongodb数据源

/**
 * 根据需求而定，给定的mongo连接信息，通过连接名称获取对应的MongoTemplate，初始化时若发现配置不全的
 * 连接信息将给定默认的MongoTemplate，通过MongoClients、url构建成mongoClient，然后通过mongoClient
 * 构建MongoTemplate 
 */
@Component
public class MongoUtils {
    private final static String URL = "url";
    private final static String DATABASE_NAME = "databaseName";
    private final static String CONNECTION_NAME = "connectionName";
    private final static String DEFAULT = "DEFAULT";
  
    private final static Map<String, MongoTemplate> mongoClient = new ConcurrentHashMap<>();
    @Resource
    private MongoTemplate mongoTemplate;

    @PostConstruct
    public void init() {
          List<Map<String, String>> datasource = new ArrayList<>();
        for (Map<String, String> connectionInfo : datasource) {
            if (connectionInfo.containsKey(URL) && connectionInfo.containsKey(DATABASE_NAME) && connectionInfo.containsKey(CONNECTION_NAME)) {
                mongoClient.put(connectionInfo.get(CONNECTION_NAME), new MongoTemplate(MongoClients.create(connectionInfo.get(URL)), connectionInfo.get(DATABASE_NAME)));
            } else {
                mongoClient.put(DEFAULT, mongoTemplate);
            }
        }
    }

    public static MongoTemplate getMongoTemplate(String connectionName) {
        return mongoClient.get(connectionName);
    }
}