data_field_dispose/data_mgo_to_tidb/save.go

package main

import "time"

func RunSaveService() {
	go SaveFunc()
	go SaveExpandFunc()
	go SaveDetailFunc()
	go SaveAttrFunc()
	go SaveIntentFunc()
	go SaveBidderFunc()
	go SaveGoodsFunc()
	//go saveErrMethod()
}

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 SaveBidderFunc() {
	arru := make([]map[string]interface{}, saveSize)
	indexu := 0
	for {
		select {
		case v := <-saveBidderPool:
			arru[indexu] = v
			indexu++
			if indexu == saveSize {
				saveBidderSp <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-saveBidderSp
					}()
					MysqlTool.InsertBulk("dwd_f_bid_package_bidder_baseinfo", BidderField, arru...)
				}(arru)
				arru = make([]map[string]interface{}, saveSize)
				indexu = 0
			}
		case <-time.After(1000 * time.Millisecond):
			if indexu > 0 {
				saveBidderSp <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-saveBidderSp
					}()
					MysqlTool.InsertBulk("dwd_f_bid_package_bidder_baseinfo", BidderField, arru...)
				}(arru[:indexu])
				arru = make([]map[string]interface{}, saveSize)
				indexu = 0
			}
		}
	}
}

func SaveGoodsFunc() {
	arru := make([]map[string]interface{}, saveSize)
	indexu := 0
	for {
		select {
		case v := <-saveGoodsPool:
			arru[indexu] = v
			indexu++
			if indexu == saveSize {
				saveGoodsSp <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-saveGoodsSp
					}()
					MysqlTool.InsertBulk("dwd_f_bid_package_goods_baseinfo", GoodsField, arru...)
				}(arru)
				arru = make([]map[string]interface{}, saveSize)
				indexu = 0
			}
		case <-time.After(1000 * time.Millisecond):
			if indexu > 0 {
				saveGoodsSp <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-saveGoodsSp
					}()
					MysqlTool.InsertBulk("dwd_f_bid_package_goods_baseinfo", GoodsField, 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
			}
		}
	}
}