123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137 |
- 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:
- }
- }
- }
|