data_field_dispose/data_mgo_to_tidb/save.go

package main

import "time"

func SaveFunc() {
	arru := make([]map[string]interface{}, saveSize)
	indexu := 0
	for {
		select {
		case v := <-saveBasePool:
			arru[indexu] = v
			indexu++
			if indexu == saveSize {
				saveBaseSp <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-saveBaseSp
					}()
					MysqlTool.InsertBulk("dwd_f_bid_baseinfo", BaseField, arru...)
				}(arru)
				arru = make([]map[string]interface{}, saveSize)
				indexu = 0
			}
		case <-time.After(1000 * time.Millisecond):
			if indexu > 0 {
				saveBaseSp <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-saveBaseSp
					}()
					MysqlTool.InsertBulk("dwd_f_bid_baseinfo", BaseField, arru...)
				}(arru[:indexu])
				arru = make([]map[string]interface{}, saveSize)
				indexu = 0
			}
		}
	}
}

func SaveExpandFunc() {
	arru := make([]map[string]interface{}, saveSize)
	indexu := 0
	for {
		select {
		case v := <-saveExpandPool:
			arru[indexu] = v
			indexu++
			if indexu == saveSize {
				saveExpandSp <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-saveExpandSp
					}()
					MysqlTool.InsertBulk("dwd_f_bid_expand_baseinfo", ExpandField, arru...)
				}(arru)
				arru = make([]map[string]interface{}, saveSize)
				indexu = 0
			}
		case <-time.After(1000 * time.Millisecond):
			if indexu > 0 {
				saveExpandSp <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-saveExpandSp
					}()
					MysqlTool.InsertBulk("dwd_f_bid_expand_baseinfo", ExpandField, arru...)
				}(arru[:indexu])
				arru = make([]map[string]interface{}, saveSize)
				indexu = 0
			}
		}
	}
}

func SaveDetailFunc() {
	arru := make([]map[string]interface{}, saveSize)
	indexu := 0
	for {
		select {
		case v := <-saveDetailPool:
			arru[indexu] = v
			indexu++
			if indexu == saveSize {
				saveDetailSp <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-saveDetailSp
					}()
					MysqlTool.InsertBulk("dwd_f_bid_detail", DetailField, arru...)
				}(arru)
				arru = make([]map[string]interface{}, saveSize)
				indexu = 0
			}
		case <-time.After(1000 * time.Millisecond):
			if indexu > 0 {
				saveDetailSp <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-saveDetailSp
					}()
					MysqlTool.InsertBulk("dwd_f_bid_detail", DetailField, arru...)
				}(arru[:indexu])
				arru = make([]map[string]interface{}, saveSize)
				indexu = 0
			}
		}
	}
}

func SaveAttrFunc() {
	arru := make([]map[string]interface{}, saveSize)
	indexu := 0
	for {
		select {
		case v := <-saveAttrPool:
			arru[indexu] = v
			indexu++
			if indexu == saveSize {
				saveAttrSp <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-saveAttrSp
					}()
					MysqlTool.InsertBulk("dwd_f_bid_file_text", AttrField, arru...)
				}(arru)
				arru = make([]map[string]interface{}, saveSize)
				indexu = 0
			}
		case <-time.After(1000 * time.Millisecond):
			if indexu > 0 {
				saveAttrSp <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-saveAttrSp
					}()
					MysqlTool.InsertBulk("dwd_f_bid_file_text", AttrField, arru...)
				}(arru[:indexu])
				arru = make([]map[string]interface{}, saveSize)
				indexu = 0
			}
		}
	}
}

func SaveIntentFunc() {
	arru := make([]map[string]interface{}, saveSize)
	indexu := 0
	for {
		select {
		case v := <-saveIntentPool:
			arru[indexu] = v
			indexu++
			if indexu == saveSize {
				saveIntentSp <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-saveIntentSp
					}()
					MysqlTool.InsertBulk("dwd_f_bid_intention_baseinfo", IntentField, arru...)
				}(arru)
				arru = make([]map[string]interface{}, saveSize)
				indexu = 0
			}
		case <-time.After(1000 * time.Millisecond):
			if indexu > 0 {
				saveIntentSp <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-saveIntentSp
					}()
					MysqlTool.InsertBulk("dwd_f_bid_intention_baseinfo", IntentField, arru...)
				}(arru[:indexu])
				arru = make([]map[string]interface{}, saveSize)
				indexu = 0
			}
		}
	}
}

func SaveProFunc() {
	arru := make([]map[string]interface{}, saveSize)
	indexu := 0
	for {
		select {
		case v := <-saveProPool:
			arru[indexu] = v
			indexu++
			if indexu == saveSize {
				saveProSp <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-saveProSp
					}()
					MysqlTool.InsertBulk("dws_f_project_baseinfo", ProField, arru...)
				}(arru)
				arru = make([]map[string]interface{}, saveSize)
				indexu = 0
			}
		case <-time.After(1000 * time.Millisecond):
			if indexu > 0 {
				saveProSp <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-saveProSp
					}()
					MysqlTool.InsertBulk("dws_f_project_baseinfo", ProField, arru...)
				}(arru[:indexu])
				arru = make([]map[string]interface{}, saveSize)
				indexu = 0
			}
		}
	}
}

func SaveProbFunc() {
	arru := make([]map[string]interface{}, saveSize)
	indexu := 0
	for {
		select {
		case v := <-saveProbPool:
			arru[indexu] = v
			indexu++
			if indexu == saveSize {
				saveProbSp <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-saveProbSp
					}()
					MysqlTool.InsertBulk("dws_f_project_business", ProBusField, arru...)
				}(arru)
				arru = make([]map[string]interface{}, saveSize)
				indexu = 0
			}
		case <-time.After(1000 * time.Millisecond):
			if indexu > 0 {
				saveProbSp <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-saveProbSp
					}()
					MysqlTool.InsertBulk("dws_f_project_business", ProBusField, arru...)
				}(arru[:indexu])
				arru = make([]map[string]interface{}, saveSize)
				indexu = 0
			}
		}
	}
}

func SaveProTagFunc() {
	arru := make([]map[string]interface{}, saveSize)
	indexu := 0
	for {
		select {
		case v := <-saveProTagPool:
			arru[indexu] = v
			indexu++
			if indexu == saveSize {
				saveProTagSp <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-saveProTagSp
					}()
					MysqlTool.InsertBulk("dws_f_project_tags", ProTagsField, arru...)
				}(arru)
				arru = make([]map[string]interface{}, saveSize)
				indexu = 0
			}
		case <-time.After(1000 * time.Millisecond):
			if indexu > 0 {
				saveProTagSp <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-saveProTagSp
					}()
					MysqlTool.InsertBulk("dws_f_project_tags", ProTagsField, arru...)
				}(arru[:indexu])
				arru = make([]map[string]interface{}, saveSize)
				indexu = 0
			}
		}
	}
}

func SaveRelationFunc() {
	arru := make([]map[string]interface{}, saveSize)
	indexu := 0
	for {
		select {
		case v := <-saveRelationPool:
			arru[indexu] = v
			indexu++
			if indexu == saveSize {
				saveRelationSp <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-saveRelationSp
					}()
					MysqlTool.InsertBulk("dws_f_bpmc_relation", RelationField, arru...)
				}(arru)
				arru = make([]map[string]interface{}, saveSize)
				indexu = 0
			}
		case <-time.After(1000 * time.Millisecond):
			if indexu > 0 {
				saveRelationSp <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-saveRelationSp
					}()
					MysqlTool.InsertBulk("dws_f_bpmc_relation", RelationField, arru...)
				}(arru[:indexu])
				arru = make([]map[string]interface{}, saveSize)
				indexu = 0
			}
		}
	}
}

// 字段错误数据
func saveErrMethod() {
	arru := make([]map[string]interface{}, 200)
	indexu := 0
	for {
		select {
		case v := <-saveErrPool:
			arru[indexu] = v
			indexu++
			if indexu == saveSize {
				saveErrSp <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-saveErrSp
					}()
					MongoB.SaveBulk("bidding_mgo_to_tidb_f_err", arru...)
				}(arru)
				arru = make([]map[string]interface{}, saveSize)
				indexu = 0
			}
		case <-time.After(1000 * time.Millisecond):
			if indexu > 0 {
				saveErrSp <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-saveErrSp
					}()
					MongoB.SaveBulk("bidding_mgo_to_tidb_f_err", arru...)
				}(arru[:indexu])
				arru = make([]map[string]interface{}, saveSize)
				indexu = 0
			}
		}
	}
}