dataprocess_file/nats.go

package main

import (
	"jygit.jydev.jianyu360.cn/BP/jynats/jnats"
	cu "jygit.jydev.jianyu360.cn/data_capture/myself_util/commonutil"
)

var (
	NatsUrl     string
	Jnats       *jnats.Jnats
	Subscribe   string
	Step        string
	NatsThreads chan bool
)

type MsgInfo struct {
	Id       string                 //消息唯一id
	CurrSetp string                 //当前步骤
	NextSetp string                 //下个步骤，特殊流程增加
	IsEnd    int                    //当前流程后结束 1-结束
	Data     map[string]interface{} //数据内容
	Extend   struct {               //有需要按照示例增加
		Extract struct { //抽取

		}
		Repeat struct { //判重
			SId string //原始id
			RId string //被替换id
		}
		MgoSave struct { //mgo保存更新
			SType string //更新u 保存s
			col   string //表
		}
		EsSave struct { //es保存更新
			SType string //更新u 保存s
			Index string //索引
		}
	}
	Err   string //错误信息 有错误会告警并终止流程
	Stime int64
	Etime int64
}

// InitNats 初始化nats
func InitNats() {
	NatsUrl = cu.ObjToString(Config["natsurl"])
	NatsThreads = make(chan bool, cu.IntAllDef(Config["natsthreads"], 10))
	Jnats = jnats.NewJnats(NatsUrl)
	Subscribe = cu.ObjToString(Config["subscribe"])
	Step = cu.ObjToString(Config["step"])
}