mongo写入链路
2022-03-21 16:51:59 0 举报主要是文件内部使用
作者其他创作
大纲/内容
1、分配后负责批量写入Mongo的线程池ThreadPoolExecutor bulkWrite2MongoScheduler2、BulkWriteQueue的拉取分配线程BulkWriteQueuePullRunnable bulkWriteConsumerThread
解析字段
初始化hms异步
更新
（省略启动步骤）初始化hms（监听元数据对应的topic）MetricHmsManager #loadConfig
AppDataMapService
MetricHmsManager #startConsume 解析数据写入限流
DmsMongoDao # updateField①、获取放入队列上下文对象②、对象放入DrainWaitLinkedBlockingQueue writeQueueLinkedBlockingQueue
调用DAO更新
BulkWriteQueuePullRunnable # writeMongo①、选择mongo库②、写入mongo库MongoCollection # bulkwrite
消费消息异步
write_mapping_error错误埋点
①、Computer#compute映射成dms的字段更新消息②、Dms#update字段校验③、Dms#send发送元数据对应的topic
初始化hms
bulk-write-error 埋点
AppDataMapConsumer
启动线程，初始化hms，去消费消息（写入mongo的消息）
bulkWriteConsumerThread异步拉取
consumer_error错误埋点
Dms.start()初始化apollo中metadata信息初始化compute实例初始化sender
DmsUpdateService # updateField①、初始化dmsCollection和rdsCollection②、初始化DmsMongoDaoImpl③、启动异步消费bulkWriteQueue④、调用mongoDao更新
写入重试队列暂时没开启
执行写入mongo
写入链路
BulkWriteQueuePullRunnable # run①、写入Mongo的bulk每批大小②、从writeQueue中获取写入对象③、bulkWrite2MongoScheduler执行提交对象
0 条评论
下一页
