jy-data-extract/udps/main.go

package main

import (
	"encoding/json"
	"flag"
	"log"
	mu "mfw/util"
	"net"
	qu "qfw/util"
	"time"

	"gopkg.in/mgo.v2/bson"
)

var udpclient mu.UdpClient //udp对象
var nextNodes []map[string]interface{}

var startDate, endDate, ip, port, stype, sid, eid string

func main() {
	//2015-11-03,2017-04-01
	//2017-04-01,2017-06-01
	//2017-06-01,2018-06-01
	//2018-06-01,2019-02-20

	/*
	ObjectId("5d767728a5cb26b9b7748868")
	ObjectId("5d77c881a5cb26b9b7de209d")  //9W条数据

	56404035af5374672e00059c
	5d4da9c8a5cb26b9b7b6bbcd  100万
	*/
	flag.StringVar(&sid, "sid", "", "开始id")
	flag.StringVar(&eid, "eid", "", "结束id")
	flag.StringVar(&startDate, "start", "", "开始日期2006-01-02")
	flag.StringVar(&endDate, "end", "", "结束日期2006-01-02")
	flag.StringVar(&ip, "ip", "127.0.0.1", "ip")
	flag.StringVar(&port, "port", "1112", "dup端口")
	flag.StringVar(&stype, "stype", "", "stype")
	flag.Parse()
	var startid, endid bson.ObjectId
	if sid != "" && eid != "" {
		startid = qu.StringTOBsonId(sid)
		endid = qu.StringTOBsonId(eid)
	} else {
		start, _ := time.ParseInLocation(qu.Date_Short_Layout, startDate, time.Local)
		end, _ := time.ParseInLocation(qu.Date_Short_Layout, endDate, time.Local)
		startid = bson.NewObjectIdWithTime(start)
		endid = bson.NewObjectIdWithTime(end)
	}
	log.Println(startid, endid, ip, port, stype)
	udpclient = mu.UdpClient{Local: ":1470", BufSize: 1024}
	udpclient.Listen(processUdpMsg)
	by, _ := json.Marshal(map[string]interface{}{
		"gtid":  startid,
		"lteid": endid,
		"stype": stype,
	})
	udpclient.WriteUdp(by, mu.OP_TYPE_DATA, &net.UDPAddr{
		IP:   net.ParseIP(ip),
		Port: qu.IntAll(port),
	})
	b := make(chan bool, 1)
	<-b
}

func processUdpMsg(act byte, data []byte, ra *net.UDPAddr) {
	switch act {
	case mu.OP_TYPE_DATA:
		var mapInfo map[string]interface{}
		err := json.Unmarshal(data, &mapInfo)
		if err != nil {
			log.Println(err)
		} else {
			log.Println(mapInfo)
		}
	case mu.OP_NOOP: //下个节点回应
		log.Println("发送成功", string(data))
	}
}