datatools/dataflow/main.go

package main

import (
	"dataflow/util"
	"fmt"
	"log"
	"time"

	"app.yhyue.com/moapp/jybase/mongodb"
	"jygit.jydev.jianyu360.cn/BP/jynats/jnats"
)

var (
	conf         util.Conf
	ThreadsLimit int
	Bidding      *mongodb.MongodbSim
	jn           *jnats.Jnats
	ProMsgInfo   = make(chan *util.MsgInfo, 50000)
	ErrMsgInfo   = make(chan *util.MsgInfo, 50000)
)

func init() {
	conf = util.GetConf()
	ThreadsLimit = conf.Config.Threads
	jn = jnats.NewJnats(conf.Config.Natsurl)
	Bidding = mongodb.NewMgoWithUser(conf.Config.Mongodb.Addr, conf.Config.Mongodb.Dbname, conf.Config.Mongodb.UserName, conf.Config.Mongodb.Password, conf.Config.Mongodb.DbSize)
	go ErrBulkSave()
}

func main() {
	step := conf.Config.Process
	subject := step.Subject
	//TODO 改造为读临时库 待确定读库模式
	//消息体
	for i := 0; i < 2; i++ {
		//消息体
		info := &util.MsgInfo{
			Id: fmt.Sprint(i),
			Data: map[string]interface{}{
				"title":  "测试消息" + fmt.Sprint(i),
				"detail": "信息源码",
			},
		}
		ProMsgInfo <- info
	}

	for i := 0; i < ThreadsLimit; i++ {
		go func() {
			for {
				info := <-ProMsgInfo
				//按照步骤发布消息
				for _, st := range step.Steps {
					newinfo, err := util.SendRequest(jn, subject, st, info, 5*time.Minute)
					if err != nil {
						log.Println("err", info.Id, err)
						reMsg(info, subject, st)                      //消息重发
						util.Send("发布消息失败告警" + info.Id + err.Error()) //告警
						info.CurrSetp = st                            //保存错误
						ErrMsgInfo <- info
						break
					} else {
						if newinfo.Err != nil {
							util.Send("消息返回错误告警" + info.Id + newinfo.Err.Error()) //告警
							info.CurrSetp = st                                    //保存错误
							ErrMsgInfo <- info
						} else {
							log.Println("消息发送成功", newinfo)
							info = newinfo
						}
					}
				}
			}
		}()
	}

	//订阅消息处理中间阶段

	select {}
}

func reMsg(msg *util.MsgInfo, subject, sts string) {
	isOk := false
	for _, st := range conf.Config.Process.Steps {
		if !isOk {
			if st == sts {
				isOk = true
				newinfo, err := util.SendRequest(jn, subject, st, msg, 5*time.Minute)
				if err != nil {
					log.Println("err", msg.Id, err)
					util.Send("消息重发失败告警" + msg.Id + err.Error()) //告警
					break
				} else {
					msg = newinfo
				}
			}
		} else {
			newinfo, err := util.SendRequest(jn, subject, st, msg, 5*time.Minute)
			if err != nil {
				log.Println("err", msg.Id, err)
				util.Send("发布消息失败告警" + msg.Id + err.Error()) //告警
				ErrMsgInfo <- msg                            //保存错误
				break
			} else {
				msg = newinfo
			}
		}
	}
}

// 错误数据保存
func ErrBulkSave() {
	elist := []map[string]interface{}{}
	for {
		select {
		case data, ok := <-ErrMsgInfo:
			if ok {
				tmp := map[string]interface{}{}
				tmp["id"] = data.Id
				tmp["currsetp"] = data.CurrSetp
				tmp["comeintime"] = time.Now().Unix()
				tmp["data"] = data.Data
				elist = append(elist, tmp)
				if len(elist) >= 20 {
					Bidding.SaveBulk("msg_err", elist...)
					elist = []map[string]interface{}{}
				}
			}
		case <-time.After(5 * time.Second): // 设置超时时间秒
			if len(elist) > 0 {
				Bidding.SaveBulk("msg_err", elist...)
				elist = []map[string]interface{}{}
			}
		default:

		}
	}
}