一种基于消息队列的MongoDB数据实时同步方法及系统

专利主权项内容

1.一种基于消息队列的MongoDB数据实时同步方法，步骤包括：创建MongoDB与Kafka之间的长连接；跟读MongoDB的日志文件oplog，对读取到的需要同步的bson数据进行初步解析，并根据oplog数据中的ns字段，将MongoDB的集合名映射成设定的ElasticSearch的索引名；将该Bson数据转码成可以缓存到Kafka的序列化数据，并缓存到Kafka上的与ElasticSearch索引名同名的Topic中；如果Kafka上不存在Topic，则自动创建Topic；从Kafka向ElasticSearch同步数据时，为每一个Topic创建一个同步线程，多线程可以并行操作；为每一个Topic创建一条从Kafka到ElasticSearch的长连接；从Kafka的Topic中读取之前缓存好的数据，并转码还原成Bson数据；解析每一条Bson数据的操作类型，调取对应的操作方法，以解析成ElasticSearch可以识别的数据结构，并缓存在本地；当本地缓存的数据达到设定的阈值时，批量提交到ElasticSearch中。