jy-data-extract/udp_winner/main.go

package main

import (
	"encoding/json"
	"fmt"
	"github.com/garyburd/redigo/redis"
	hisRedis "github.com/go-redis/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, BuyerFields, AgencyFields     []string
	SourceClient, FClient                 *MongodbSim
	RedisPool                             redis.Pool
	HisRedisPool                          *hisRedis.Client
	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
	CPool                                 chan bool
)

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

	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"}
	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()
	HisRedisPool = hisRedis.NewClient(&hisRedis.Options{
		Addr:         "127.0.0.1:6380",
		DB:           1,
		DialTimeout:  10 * time.Second,
		ReadTimeout:  30 * time.Second,
		WriteTimeout: 30 * time.Second,
		PoolSize:     30,
		MinIdleConns: 20,
		PoolTimeout:  30 * time.Second,
	})
}

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() //定时任务
	go TimedTaskBuyer()  //定时任务
	go TimedTaskAgency() //定时任务
	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 {
				//阻塞
				CPool <- true
				if key == "winner" {
					go TaskWinner(tmp)
				} else if key == "buyer" {
					go TaskBuyer(tmp)
				} else if key == "agency" {
					go TaskAgency(tmp)
				}
			}
		}
	case mu.OP_NOOP: //下个节点回应
		log.Println("发送成功", string(data))
	}
}