DataService/jy_publishing/main.go

package main

import (
	"encoding/json"
	"fmt"
	"github.com/zeromicro/go-zero/core/discov"
	"github.com/zeromicro/go-zero/zrpc"
	ms "jy_publishing/megaloscope"
	nsq "jy_publishing/nsq"
	"net"
	"strings"
	"utils"
	"utils/elastic"
	"utils/log"
	"utils/mongodb"
)

var (
	Sysconfig         map[string]interface{}
	MgoBid, MgoExt    *mongodb.MongodbSim
	BidColl           string
	ExtColl, ExtColl1 string
	Es                *elastic.Elastic
	Index, IndexAll   string
	Itype             string
	JyRpcClient       zrpc.Client
	ClientAddr        string
	MCJy, MCAtts      *nsq.Consumer
	MProducer         *nsq.Producer
	Ms                *ms.Megaloscope //敏感词
	UdpClient         util.UdpClient  //udp对象
)

func init() {
	log.InitLogger("./log/All.log", "debug")
	util.ReadConfig(&Sysconfig)
	bidding := Sysconfig["bidding"].(map[string]interface{})
	BidColl = bidding["dbColl"].(string)
	MgoBid = &mongodb.MongodbSim{
		MongodbAddr: bidding["addr"].(string),
		Size:        util.IntAllDef(Sysconfig["mgoPoolSize"], 5),
		DbName:      bidding["dbName"].(string),
		UserName:    bidding["uname"].(string),
		Password:    bidding["upwd"].(string),
	}
	MgoBid.InitPool()
	extract := Sysconfig["extract"].(map[string]interface{})
	ExtColl = extract["dbColl"].(string)
	ExtColl1 = extract["dbColl1"].(string)
	MgoExt = &mongodb.MongodbSim{
		MongodbAddr: extract["addr"].(string),
		Size:        util.IntAllDef(Sysconfig["mgoPoolSize"], 5),
		DbName:      extract["dbName"].(string),
	}
	MgoExt.InitPool()
	es := Sysconfig["es"].(map[string]interface{})
	Es = &elastic.Elastic{
		S_esurl: util.ObjToString(es["addr"]),
		I_size:  util.IntAllDef(es["pool"], 10),
	}
	Index = util.ObjToString(es["index"])
	IndexAll = util.ObjToString(es["bidding_all"])
	Itype = util.ObjToString(es["itype"])
	Es.InitElasticSize()

	//加载敏感词文件
	Ms = ms.NewMegaloscope("./rules.txt")

	jyRpc := Sysconfig["jy_rpc"].(map[string]interface{})
	JyRpcClient = zrpc.MustNewClient(zrpc.RpcClientConf{
		Etcd: discov.EtcdConf{
			Hosts: strings.Split(util.ObjToString(jyRpc["addr"]), ","),
			Key:   util.ObjToString(jyRpc["key"]),
		},
	})

	//initUdp()
	//initNsq()
}

func initNsq() {
	cof := Sysconfig["nsq_jy"].(map[string]interface{})
	MCJy, _ = nsq.NewConsumer(&nsq.Cconfig{
		IsJsonEncode: true, //与生产者配置对应，设为true会取第1个字节进行类型判断
		Addr:         util.ObjToString(cof["addr"]),
		ConnectType:  0, //默认连接nsqd
		Topic:        util.ObjToString(cof["topic"]),
		Channel:      util.ObjToString(cof["channel"]),
		Concurrent:   util.IntAllDef(cof["concurrent"], 1), //并发数
	})
	cofAtts := Sysconfig["nsq_attachment"].(map[string]interface{})
	MProducer, _ = nsq.NewProducer(util.ObjToString(cofAtts["addr"]), util.ObjToString(cofAtts["topic"]), true)
	MCAtts, _ = nsq.NewConsumer(&nsq.Cconfig{
		IsJsonEncode: false, //与生产者配置对应，设为true会取第1个字节进行类型判断
		Addr:         util.ObjToString(cofAtts["addr"]),
		ConnectType:  0, //默认连接nsqd
		Topic:        util.ObjToString(cofAtts["topic-result"]),
		Channel:      util.ObjToString(cofAtts["channel"]),
		Concurrent:   util.IntAllDef(cofAtts["concurrent"], 1), //并发数
	})
}

func initUdp() {
	updport := Sysconfig["udpPort"].(string)
	UdpClient = util.UdpClient{Local: updport, BufSize: 1024}
	log.Info("Udp 监听 port: " + updport)
	UdpClient.Listen(processUdpMsg)
}

func main() {
	//go jyNsqMethod()
	go attsNsqMethod()

	attsMap := map[string]interface{}{"1": map[string]interface{}{
		"fid":      "1e0828202d4269d006f95e10e8f7afea4794a3d066d687f160f7dcafba02f74c.docx",
		"filename": "附件202110121614016176.docx",
		"ftype":    "docx",
	}}
	other := map[string]interface{}{
		"id":      "113",
		"action":  "1",
		"msgType": "1",
		"title":   []string{"汉"},
		"detail":  []string{},
	}
	//otherJson, _ := json.Marshal(other)
	//var attsArr []*pb.Request
	var attsArr []map[string]interface{}
	for _, m := range attsMap {
		m1 := m.(map[string]interface{})
		//attsArr = append(attsArr, &pb.Request{
		//	FileUrl:     util.ObjToString(m1["fid"]),
		//	FileName:    util.ObjToString(m1["filename"]),
		//	FileType:    util.ObjToString(m1["ftype"]),
		//	ReturnType:  1,
		//	ExtractType: 1,
		//})
		attsArr = append(attsArr, map[string]interface{}{
			"fileUrl":     util.ObjToString(m1["fid"]),
			"fileName":    util.ObjToString(m1["filename"]),
			"fileType":    util.ObjToString(m1["ftype"]),
			"returnType":  1,
			"extractType": 0,
		})
	}
	msginfo := map[string]interface{}{
		"message": attsArr,
		"other":   other,
		"topic":   "attachment-txt",
	}
	util.Debug(msginfo)
	_ = MProducer.Publish(msginfo)

	c := make(chan bool, 1)
	<-c
}

// @Description 剑鱼消息队列 按照类型处理消息
// @Author J 2022/4/14 11:42 AM
func jyNsqMethod() {
	for {
		select {
		case obj := <-MCJy.Ch: //从通道读取即可
			fmt.Println("---", obj)
			taskInfo(obj)
		}
	}
}

// @Description 附件处理完成的数据
// @Author J 2022/4/14 1:18 PM
func attsNsqMethod() {
	for {
		select {
		case obj := <-MCAtts.Ch:
			var resp *AttsResponse
			_ = json.Unmarshal(obj.([]byte), &resp)
			fmt.Println("---", resp)
			taskAtts(resp)
		}
	}
}

func processUdpMsg(act byte, data []byte, ra *net.UDPAddr) {
	switch act {
	case util.OP_TYPE_DATA: //测试接收
		var resp map[string]interface{}
		err := json.Unmarshal(data, &resp)
		log.Info("udp receive...", log.Field("data", resp))
		if err != nil {
			//go Udpclient.WriteUdp([]byte{}, mu.OP_NOOP, ra) //回应上一个节点
		} else if resp != nil {
			key, _ := resp["key"].(string)
			if key == "" {
				key = "udpok"
			}
			go UdpClient.WriteUdp([]byte(key), util.OP_NOOP, ra)
			tasktype, _ := resp["stype"].(string)
			switch tasktype {
			case "jyfb_data_over":
				go func() {
					DelMethod(resp["infoid"].(string))
				}()
			}
		}
	case util.OP_NOOP: //下个节点回应
		log.Info("接收回应：", log.Field("data", string(data)))
	}

}