package main import ( "encoding/json" "fmt" "github.com/garyburd/redigo/redis" "go.mongodb.org/mongo-driver/bson" es "gopkg.in/olivere/elastic.v1" "log" mu "mfw/util" "net" "qfw/common/src/qfw/util/elastic" "qfw/util" "regexp" "strconv" "strings" "time" ) var ( Config = make(map[string]string) Fields []string SourceClient, FClient *MongodbSim RedisPool redis.Pool Addrs = make(map[string]interface{}, 0) //省市县 udpclient mu.UdpClient //udp对象 ElasticClientIndex, ElasticClientType string Reg_xing = regexp.MustCompile(`\*{1,}`) Reg_person = regexp.MustCompile("[\u4E00-\u9FA5\\s]+") Reg_tel = regexp.MustCompile(`^[0-9\-\s]*$`) EsConn *es.Client Updport int ) /** 新增 初始化 */ func init() { log.SetFlags(log.Ldate | log.Ltime | log.Lshortfile) util.ReadConfig(&Config) log.Println(Config) Fields = []string{"_id", "contact", "partners", "business_scope", "company_address", "capital", "establish_date", "legal_person", "company_type", "district", "city", "province", "area_code", "credit_no", "company_name", "history_name", "topscopeclass", "wechat_accounts", "alias", "website", "report_websites"} var err error pool_size, _ := strconv.Atoi(Config["pool_size"]) //mongo init SourceClient = new(MongodbSim) SourceClient.MongodbAddr = Config["mgoinit"] SourceClient.Size = pool_size //mongodbSim.DbName = "qfw" SourceClient.InitPool() SourceClientmgoConn := SourceClient.GetMgoConn() defer SourceClient.DestoryMongoConn(SourceClientmgoConn) FClient = new(MongodbSim) FClient.MongodbAddr = Config["mgourl"] FClient.Size = pool_size FClient.DbName = Config["mgodb_extract_kf"] //mongodbSim.DbName = "qfw" FClient.InitPool() FClientmgoConn := FClient.GetMgoConn() defer FClient.DestoryMongoConn(FClientmgoConn) //加载省市县代码 cursor2 := FClientmgoConn.DB(Config["mgodb_extract_kf"]).C("address").Find(bson.M{}).Select(bson.M{"province": 1, "code": 1, "city": 1, "district": 1}).Iter() //defer FClient.Connect(cc) if cursor2 == nil { log.Fatalln(cursor2) } tmp := make(map[string]interface{}) for cursor2.Next(&tmp) { code := tmp["code"] if code != nil && strings.TrimSpace(code.(string)) != "" { Addrs[fmt.Sprint(code)] = tmp } } log.Println(len(Addrs)) //es.NewClient(es.SetURL(addrs...), es.SetMaxRetries(2), es.SetSniff(false)) //es init elastic.InitElasticSize(Config["elasticsearch"], 50) EsConn = elastic.GetEsConn() defer elastic.DestoryEsConn(EsConn) //redis RedisPool = redis.Pool{ MaxIdle: 50, IdleTimeout: 10 * time.Second, Dial: func() (redis.Conn, error) { conn, e := redis.Dial("tcp", Config["redis"]) if e != nil { return conn, e } _, err = conn.Do("SELECT", "1") if err != nil { return nil, err } return conn, nil }} c := RedisPool.Get() if _, err := c.Do("PING"); err != nil { log.Fatalln("redis err:", err) } c.Close() } func main() { //udp updport := Config["udpport"] Updport, _ = strconv.Atoi(Config["port"]) udpclient = mu.UdpClient{Local: updport, BufSize: 1024} udpclient.Listen(processUdpMsg) log.Println("Udp服务监听", updport) log.Println("发送端口port:", Updport) go TimedTaskWinner() //定时任务 c := make(chan int, 1) <-c } func processUdpMsg(act byte, data []byte, ra *net.UDPAddr) { log.Println(act, string(data), ra) switch act { case mu.OP_TYPE_DATA: //上个节点的数据 //从表中开始处理生成企业数据 tmp := new(map[string]interface{}) err := json.Unmarshal(data, &tmp) if err != nil { log.Println("err:", err) udpclient.WriteUdp([]byte("err:"+err.Error()), mu.OP_NOOP, ra) return } else if tmp != nil { if key, ok := (*tmp)["key"].(string); ok { udpclient.WriteUdp([]byte(key), mu.OP_NOOP, ra) } else { udpclient.WriteUdp([]byte("udpok"), mu.OP_NOOP, ra) } //data_type:winner data_type:buyer data_type:agency //data_info:save//存量 data_info:add //增量 if key, ok := (*tmp)["data_type"].(string); ok { if key == "winner" { go TaskWinner(tmp) } else if key == "buyer" { } else if key == "agency" { } } } case mu.OP_NOOP: //下个节点回应 log.Println("发送成功", string(data)) } }