jy-data-extract/udp_winner/main.go

package main

import (
	"encoding/json"
	"fmt"
	"log"
	mu "mfw/util"
	"net"
	"qfw/util"
	"qfw/common/src/qfw/util/mongodb"
	"regexp"
	"strconv"
	"strings"
	"time"

	"github.com/garyburd/redigo/redis"
	hisRedis "github.com/go-redis/redis"
	"gopkg.in/mgo.v2/bson"
)

var (
	Config                               = make(map[string]string)
	Sysconfig                            map[string]interface{}
	Fields, BuyerFields, AgencyFields    []string
	SourceClient, FClient                *mongodb.MongodbSim
	RedisPool                            redis.Pool
	HisRedisPool                         *hisRedis.Client
	Addrs                                = make(map[string]interface{}, 0) //省市县
	udpclient                            mu.UdpClient                      //udp对象
	Reg_person                           = regexp.MustCompile("[\u4e00-\u9fa5]+")
	Reg_xing                             = regexp.MustCompile(`\*{1,}`)
	Reg_tel                              = regexp.MustCompile(`^[0-9\-\s]*$`)
	Updport                              int
	CPoolWinner, CPoolBuery, CPoolAgency chan bool
	//his_redis db
	redis_winner_db, redis_buyer_db, redis_agency_db int
	//异常表正则匹配处理
	WinnerRegOk, WinnerRegErr, AgencyRegOk, AgencyRegErr, BuyerRegOk, BuyerRegErr []regexp.Regexp
)

/**
新增
初始化
*/
func init() {
	log.SetFlags(log.Ldate | log.Ltime | log.Lshortfile)
	util.ReadConfig(&Config)
	util.ReadConfig(&Sysconfig)
	log.Println(Sysconfig)
	var err error
	cpnum, err := strconv.Atoi(Config["chan_pool_num"])
	if err != nil {
		log.Fatalln(err)
	}
	CPoolWinner = make(chan bool, cpnum)
	CPoolBuery = make(chan bool, cpnum)
	CPoolAgency = make(chan bool, cpnum)
	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", "wechat_accounts",
		"alias", "website", "report_websites", "industry"}

	BuyerFields = []string{"_id", "contact", "type", "ranks", "buyerclass",
		"address", "district", "city", "province", "area_code", "credit_no", "buyer_name",
		"history_name", "wechat_accounts", "website", "report_websites"}

	AgencyFields = []string{"_id", "contact", "type", "ranks",
		"address", "district", "city", "province", "area_code", "credit_no", "agency_name",
		"history_name", "wechat_accounts", "website", "report_websites"}

	initRdis()
	initMongo()
	initReg()
	Updport, err = strconv.Atoi(Config["port"])
	if err != nil{
		log.Fatalln(err)
	}
}

func main() {
	//udp
	updport := Config["udpport"]
	udpclient = mu.UdpClient{Local: updport, BufSize: 1024}
	udpclient.Listen(processUdpMsg)
	log.Println("Udp服务监听", updport)
	go TimedTaskWinner() //定时任务
	go TimedTaskBuyer()  //定时任务
	go TimedTaskAgency() //定时任务
	go checkMapJob()
	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_info:save//存量   data_info:add //增量
			//阻塞
			CPoolWinner <- true
			go func(mapinfo *map[string]interface{}) {
				defer func() { <-CPoolWinner }()
				TaskWinner(mapinfo)
			}(tmp)
			CPoolBuery <- true
			go func(mapinfo *map[string]interface{}) {
				defer func() { <-CPoolBuery }()
				TaskBuyer(mapinfo)
			}(tmp)
			CPoolAgency <- true
			go func(mapinfo *map[string]interface{}) {
				defer func() { <-CPoolAgency }()
				TaskAgency(mapinfo)
			}(tmp)

		}
	case mu.OP_NOOP: //下个节点回应
		log.Println("发送成功", string(data))
	}
}

func initRdis() {
	var err error
	//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()
	HisRedisPool = hisRedis.NewClient(&hisRedis.Options{
		Addr:         Config["his_redis"],
		DB:           1,
		DialTimeout:  10 * time.Second,
		ReadTimeout:  30 * time.Second,
		WriteTimeout: 30 * time.Second,
		PoolSize:     30,
		MinIdleConns: 20,
		PoolTimeout:  30 * time.Second,
	})
	redis_winner_db, _ = strconv.Atoi(Config["redis_winner_db"])
	redis_buyer_db, _ = strconv.Atoi(Config["redis_buyer_db"])
	redis_agency_db, _ = strconv.Atoi(Config["redis_agency_db"])
}

func initMongo() {
	//mongo init
	pool_size, _ := strconv.Atoi(Config["pool_size"])
	SourceClient = new(mongodb.MongodbSim)
	SourceClient.MongodbAddr = Config["mgoinit"]
	SourceClient.Size = pool_size
	SourceClient.DbName = Config["mgodb_bidding"]
	SourceClient.InitPool()

	FClient = new(mongodb.MongodbSim)
	FClient.MongodbAddr = Config["mgourl"]
	FClient.Size = pool_size
	FClient.DbName = Config["mgodb_extract_kf"]
	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))
}

func initReg() {
	FClientmgoConnReg := FClient.GetMgoConn()
	defer FClient.DestoryMongoConn(FClientmgoConnReg)
	findReg, b := FClient.Find(Config["mgo_qyk_reg"], bson.M{"delete": false, "isuse": true}, bson.M{"_id": 1}, nil, false, -1, 0)
	if !b {
		log.Fatalln("查询正则失败")
	}
	for _, v := range *findReg {
		s_field, ok := v["s_field"].(string) //字段
		s_rule, ok2 := v["s_rule"].(string)  //正则
		s_type, ok3 := v["s_type"].(string)  //ok or err
		if !ok || !ok2 || !ok3 || s_field == "" || s_rule == "" || s_type == "" {
			continue
		}
		var pattern string
		if strings.Contains(s_rule, "\\u") {
			s_rule = strings.Replace(s_rule, "\\", "\\\\", -1)
			s_rule = strings.Replace(s_rule, "\\\\u", "\\u", -1)
			pattern, _ = strconv.Unquote(`"` + s_rule + `"`)
		} else {
			pattern = s_rule
		}
		regtmp := regexp.MustCompile(pattern)
		if s_field == "winner" {
			if s_type == "ok" {
				WinnerRegOk = append(WinnerRegOk, *regtmp)
			} else if s_type == "err" {
				WinnerRegErr = append(WinnerRegErr, *regtmp)
			}
		} else if s_field == "buyer" {
			if s_type == "ok" {
				BuyerRegOk = append(BuyerRegOk, *regtmp)
			} else if s_type == "err" {
				BuyerRegErr = append(BuyerRegErr, *regtmp)
			}
		} else if s_field == "agency" {
			if s_type == "ok" {
				AgencyRegOk = append(AgencyRegOk, *regtmp)
			} else if s_type == "err" {
				AgencyRegErr = append(AgencyRegErr, *regtmp)
			}
		}
	}
	log.Println(len(WinnerRegOk), len(WinnerRegErr), len(BuyerRegOk), len(BuyerRegErr), len(AgencyRegOk), len(AgencyRegErr))
}

//通知下个节点nextNode
func nextNode(stype string, updatatime int64) {
	to, _ := Sysconfig["nextNode"].(map[string]interface{})
	log.Println(stype, to)
	key := stype + "-" + fmt.Sprint(updatatime)
	by, _ := json.Marshal(map[string]interface{}{
		"query": map[string]interface{}{
			"updatatime": map[string]interface{}{
				"$gte": updatatime,
			},
		},
		"stype": stype,
		"key":   key,
	})
	addr := &net.UDPAddr{
		IP:   net.ParseIP(to["addr"].(string)),
		Port: util.IntAll(to["port"]),
	}
	node := &udpNode{by, addr, time.Now().Unix(), 0}
	udptaskmap.Store(key, node)
	udpclient.WriteUdp(by, mu.OP_TYPE_DATA, addr)
}