data_field_dispose/processing_ids/main.go

package main

import (
	"app.yhyue.com/data_processing/common_utils/log"
	"app.yhyue.com/data_processing/common_utils/mongodb"
	"fmt"
	"github.com/robfig/cron"
	"go.uber.org/zap"
	"io/ioutil"
	"net/http"
	"time"
)

var (
	MgoBid          *mongodb.MongodbSim
	lastId, startId string

	mail_to  = "zhangjinkun@topnet.net.cn,wangjianghan@topnet.net.cn,maxiaoshan@topnet.net.cn,zhengkun@topnet.net.cn"
	mail_api = "http://172.17.145.179:19281/_send/_mail"
)

func init() {
	MgoBid = &mongodb.MongodbSim{
		MongodbAddr: "172.17.4.187:27082,172.17.145.163:27083",
		DbName:      "qfw",
		Size:        5,
		UserName:    "SJZY_RWESBid_Other",
		Password:    "SJZY@O17t8herB3B",
	}
	MgoBid.InitPool()

	startId = ""
}

func main() {
	TimeTask()
}

func TimeTask() {
	c := cron.New()
	cronstr := "0 */5 * * * ?" // 每5min执行一次
	_ = c.AddFunc(cronstr, func() {
		taskinfo()
	})
	c.Start()
}

func taskinfo() {
	info, _ := MgoBid.Find("bidding", nil, `{"_id": -1}`, `{"_id": 1}`, true, -1, -1)
	if info != nil {
		lastId = mongodb.BsonIdToSId((*info)[0]["_id"])
	} else {
		sendMail("bidding表id查询失败")
		return
	}
	q := map[string]interface{}{"$gt": mongodb.StringTOBsonId(startId), "$lte": mongodb.StringTOBsonId(lastId)}
	count := MgoBid.Count("bidding", q)
	ids := fmt.Sprintf("%s-%s", startId, lastId)
	if count <= 0 {
		sendMail(fmt.Sprintf("bidding表id段数据查询失败，%s", ids))
		return
	}
	save := make(map[string]interface{})
	save["gtid"] = startId
	save["lteid"] = lastId
	save["count"] = count
	now := time.Now().Unix()
	save["dataprocess"] = 0
	save["createtime"] = now
	save["updatetime"] = now
	MgoBid.Save("bidding_processing_ids", save)

}

func sendMail(content string) {
	res, err := http.Get(fmt.Sprintf("%s?to=%s&title=%s&body=%s", mail_api, mail_api, "processing_ids-send-fail", content))
	if err == nil {
		defer res.Body.Close()
		read, err := ioutil.ReadAll(res.Body)
		log.Info("send mail ...", zap.String("r:", string(read)), zap.Any("err:", err))
	}
}