spider_task/src/timetask/listdatamove.go

package timetask

import (
	"github.com/donnie4w/go-logger/logger"
	qu "qfw/util"
	"sync"
	"time"
	"util"
)

//列表页历史数据迁移
func MoveListData() {
	logger.Debug("开始迁移spider_highlistdata数据...")
	//1、spider_highlistdata数据迁移，只保留一个月
	defer qu.Catch()
	sess := util.MgoS.GetMgoConn()
	defer util.MgoS.DestoryMongoConn(sess)
	query := map[string]interface{}{
		"comeintime": map[string]interface{}{
			"$lt": time.Now().Unix() - 30*86400,
		},
	}
	lock := &sync.Mutex{}
	wg := &sync.WaitGroup{}
	ch := make(chan bool, 3)
	arr := []map[string]interface{}{}
	count, _ := sess.DB(util.MgoS.DbName).C("spider_highlistdata").Find(&query).Count()
	logger.Debug("spider_highlistdata迁移数据量：", count)
	it := sess.DB(util.MgoS.DbName).C("spider_highlistdata").Find(&query).Iter()
	n := 0
	for tmp := make(map[string]interface{}); it.Next(&tmp); n++ {
		wg.Add(1)
		ch <- true
		go func(tmp map[string]interface{}) {
			defer func() {
				<-ch
				wg.Done()
			}()
			lock.Lock()
			arr = append(arr, tmp)
			if len(arr) > 500 {
				util.MgoS.SaveBulk("spider_highlistdata_back", arr...)
				arr = []map[string]interface{}{}
			}
			lock.Unlock()
		}(tmp)
		if n%1000 == 0 {
			logger.Debug(n)
		}
		tmp = map[string]interface{}{}
	}
	wg.Wait()
	lock.Lock()
	if len(arr) > 0 {
		util.MgoS.SaveBulk("spider_highlistdata_back", arr...)
		arr = []map[string]interface{}{}
	}
	lock.Unlock()
	//删除原表中数据
	delNum := util.MgoS.Delete("spider_highlistdata", query)
	logger.Debug("spider_highlistdata删除数据量：", delNum)

	logger.Debug("开始迁移spider_listdata数据...")
	//1、spider_listdata数据迁移，只保留两个月
	query["comeintime"] = map[string]interface{}{
		"$lt": time.Now().Unix() - 60*86400,
	}
	count, _ = sess.DB(util.MgoS.DbName).C("spider_listdata").Find(&query).Count()
	logger.Debug("spider_listdata迁移数据量：", count)
	it1 := sess.DB(util.MgoS.DbName).C("spider_listdata").Find(&query).Iter()
	n1 := 0
	for tmp := make(map[string]interface{}); it1.Next(&tmp); n1++ {
		wg.Add(1)
		ch <- true
		go func(tmp map[string]interface{}) {
			defer func() {
				<-ch
				wg.Done()
			}()
			lock.Lock()
			arr = append(arr, tmp)
			if len(arr) > 500 {
				util.MgoS.SaveBulk("spider_listdata_back", arr...)
				arr = []map[string]interface{}{}
			}
			lock.Unlock()
		}(tmp)
		if n1%1000 == 0 {
			logger.Debug(n1)
		}
		tmp = map[string]interface{}{}
	}
	wg.Wait()
	lock.Lock()
	if len(arr) > 0 {
		util.MgoS.SaveBulk("spider_listdata_back", arr...)
		arr = []map[string]interface{}{}
	}
	lock.Unlock()
	//删除原表中数据
	delNum = util.MgoS.Delete("spider_listdata", query)
	logger.Debug("spider_listdata删除数据量：", delNum)
	logger.Debug("数据迁移完成...")
}