便签库消费者rbmq

README.md

@@ -32,3 +32,9 @@ example:
 
 
 
+
+
+go run .\erp\main.go -configDir=.\config\  -logDir=.\log\
+
+
+

dal/db/db.go

@@ -6,6 +6,8 @@ import (
 
 var (
 	DB *sqlx.DB
+	LabelDB *sqlx.DB
+	CmsDB *sqlx.DB
 )
 
 func Init(dns string) error {
@@ -19,7 +21,52 @@ func Init(dns string) error {
 	if err != nil {
 		return err
 	}
-	DB.SetMaxOpenConns(100)
-	DB.SetMaxIdleConns(16)
+	DB.SetMaxOpenConns(50)
+	DB.SetMaxIdleConns(10)
 	return nil
 }
+
+func InitLabelDB(dns string) error{
+	var err error
+	LabelDB, err = sqlx.Open("mysql", dns)
+	if err != nil {
+		return err
+	}
+
+	err = LabelDB.Ping()
+	if err != nil {
+		return err
+	}
+	LabelDB.SetMaxOpenConns(50)
+	LabelDB.SetMaxIdleConns(10)
+	return nil
+}
+func InitCmslDB(dns string) error{
+	var err error
+	CmsDB, err = sqlx.Open("mysql", dns)
+	if err != nil {
+		return err
+	}
+
+	err = CmsDB.Ping()
+	if err != nil {
+		return err
+	}
+	CmsDB.SetMaxOpenConns(50)
+	CmsDB.SetMaxIdleConns(10)
+	return nil
+}
+
+func CreateDB(action string,dns string) error{
+	var err error
+	switch action {
+		case "Init":
+			err = Init(dns)
+		case "InitLabelDB":
+			err = InitLabelDB(dns)
+		case "InitCmsDB":
+		err = InitCmslDB(dns)
+	}
+	return err
+}
+

dal/label/label.go

+package label
+
+import (
+	"errors"
+	"fmt"
+	"go-queue-server/dal/db"
+	"time"
+)
+
+
+type Entrys struct {
+	TemplateId int `json:"templateId"`
+	Sale_order_sn string `json:"sale_order_sn"`
+	Goods_type string `json:"goods_type"`
+	Goods_brand string `json:"goods_brand"`
+	Goods_number int `json:"goods_number"`
+	Sku_code string `json:"sku_code"`
+	Customer_code string `json:"customer_code"`
+	Customer_type string `json:"customer_type"`
+	Customer_order  string `json:"customer_order"`
+}
+
+
+type TemplateRelation struct {
+	CreateUser string `json:"createUser"`
+	Customer_com string `json:"customer_com"`
+	ENTRYS *[]Entrys `json:"ENTRYS"`
+}
+
+type SaleOrderLists struct {
+	Id string `json:"id" db:"id"`
+	Sale_order_sn string `json:"sale_order_sn" db:"sale_order_sn"`
+	Goods_type string `json:"goods_type" db:"goods_type"`
+	Goods_brand string `json:"goods_brand" db:"goods_brand"`
+	Goods_number string `json:"goods_number" db:"goods_number"`
+	Sku_code string `json:"sku_code" db:"sku_code"`
+	Customer_code string `json:"customer_code" db:"customer_code"`
+	Customer_type string `json:"customer_type" db:"customer_type"`
+	Customer_com string `json:"customer_com" db:"customer_com"`
+	CreateUser string `json:"createUser" db:"createUser"`
+	Create_time int `json:"create_time" db:"create_time"`
+}
+
+type UserInfo struct {
+	UserId int `db:"userId"`
+}
+
+func AddTmplRelations(relations TemplateRelation) error{
+	//saleOrderLists := []*SaleOrderLists{}
+	//db.LabelDB.Select(&saleOrderLists,"select *  from lie_sale_order_lists")
+	fmt.Println(relations)
+
+	var(
+		err error
+		sale_order_lists_id int64
+	)
+	//找出销售员对应后台登陆的用户id
+	userInfo := UserInfo{}
+	err = db.CmsDB.Get(&userInfo,"select userId from user_info where name = ? ",relations.CreateUser);
+	if err != nil{
+		return errors.New(fmt.Sprintf("没找到对应的销售员: %s ",relations.CreateUser))
+
+	}
+	tx, _ := db.LabelDB.Beginx()
+
+
+	for _,item := range *relations.ENTRYS{
+
+		insertSql1 := `insert into lie_sale_order_lists(sale_order_sn,goods_type,goods_brand,goods_number,sku_code,customer_code,customer_type,customer_com,createUser,create_time) values (?,?,?,?,?,?,?,?,?,?)`
+		result, err := tx.Exec(insertSql1,item.Sale_order_sn,item.Goods_type,item.Goods_brand,item.Goods_number,item.Sku_code,item.Customer_code,item.Customer_type,relations.Customer_com,relations.CreateUser,time.Now().Unix())
+		if err != nil{
+			break
+		}
+
+		sale_order_lists_id,err = result.LastInsertId()
+		if(err != nil){
+			break
+		}
+
+
+
+		insertSql2 := `insert into lie_tmpl_relations (template_id,type_id,type,create_time,create_userid,create_username) values (?,?,?,?,?,?)`
+		_, err = tx.Exec(insertSql2,item.TemplateId,sale_order_lists_id,1,time.Now().Unix(),userInfo.UserId,relations.CreateUser)
+		if(err != nil){
+			break
+		}
+
+	}
+	if err != nil{
+		tx.Rollback()
+		return err
+	}
+
+	err = tx.Commit()
+
+	return err
+
+}

erp/label/label.go

+package label
+
+import (
+	"encoding/json"
+	"fmt"
+	"github.com/ichunt2019/logger"
+	"go-queue-server/dal/label"
+)
+
+
+
+
+
+/*
+添加销售单到标签模板
+ */
+func AddTemplateRelations(erpData []byte) (err error){
+	templateRelations := label.TemplateRelation{}
+	fmt.Println(string(erpData))
+	err = json.Unmarshal(erpData,&templateRelations )
+	if err != nil{
+		logger.Error("解析erp传输的销售单json报错: %s", err)
+		return err
+	}
+	err = label.AddTmplRelations(templateRelations)
+	if err != nil{
+		logger.Error("插入销售单数据错误 %s", err)
+	}
+	return err
+}

erp/main.go

+package main
+
+import (
+	"flag"
+	_ "fmt"
+	_ "github.com/ichunt2019/go-msgserver/utils/rabbitmq"
+	"github.com/ichunt2019/go-rabbitmq/utils/rabbitmq"
+	"github.com/ichunt2019/logger"
+	"go-queue-server/dal/db"
+	"go-queue-server/erp/label"
+	"go-queue-server/util"
+	_ "sync"
+	_ "time"
+)
+
+
+
+
+
+type RecvPro struct {
+
+}
+
+//// 实现消费者 消费消息失败 自动进入延时尝试  尝试3次之后入库db
+func (t *RecvPro) Consumer(dataByte []byte) error {
+	logger.Info(string(dataByte))
+	error := label.AddTemplateRelations(dataByte)
+	return error
+}
+
+//消息已经消费3次 失败了 请进行处理
+func (t *RecvPro) FailAction(dataByte []byte) error {
+	logger.Error("任务处理失败了，我要进入db日志库了")
+	logger.Error("任务处理失败了，发送钉钉消息通知主人")
+	logger.Error(string(dataByte))
+	return nil
+}
+
+
+func initDb(action string,dns string) (err error) {
+	err = db.CreateDB(action,dns)
+	if err != nil {
+		return
+	}
+
+	return
+}
+
+var ConfigDir string
+var LogDir string
+
+// 解析命令行参数
+func initArgs() {
+	// worker -config ./worker.json
+	// worker -h
+	flag.StringVar(&ConfigDir, "configDir", "", "配置文件")
+	flag.StringVar(&LogDir, "logDir", "", "日志目录")
+	flag.Parse()
+}
+
+
+func main() {
+
+	initArgs()
+
+	//初始化配置文件
+	util.Init(ConfigDir)
+
+	//
+	logConfig := make(map[string]string)
+	logConfig["log_path"] = LogDir+"erp/label"
+	logConfig["log_chan_size"] = "1000"
+	logger.InitLogger("file",logConfig)
+	logger.Init()
+	//初始化db
+	initDb("Init",util.Configs.Liexin_databases.Dns)
+	initDb("InitLabelDB",util.Configs.LiexinLabel_databases.Dns)
+	initDb("InitCmsDB",util.Configs.Liexincms_databases.Dns)
+
+	t := &RecvPro{}
+
+
+	rabbitmq.Recv(rabbitmq.QueueExchange{
+		util.Configs.Rabbitmq_erp_label.QueueName,
+		util.Configs.Rabbitmq_erp_label.RoutingKey,
+		util.Configs.Rabbitmq_erp_label.Exchange,
+		util.Configs.Rabbitmq_erp_label.Type,
+		util.Configs.Rabbitmq_erp_label.Dns,
+	},t,3)
+}
\ No newline at end of file

order/sendPurchaseEmail/main.go

@@ -116,6 +116,11 @@ func main() {
 	//	time.Sleep(time.Microsecond*10)
 	//}
 
+	/*
+
+	{"data":[{"orderNum":"ZTY-HK202004150001","goods":"RC0402JR-070RL","purchasingPrice":"0.001300","unitPrice":"0.001300","brand":"Yageo","nums":"10472","currency":"\u7f8e\u5143","totalPrice":"13.61","remarks":"xxx","material":"E6nwafP+T1+NIcwhjrlPh4iIKlg=","purchasing":"\u5e73\u53f0","projectId":1,"deliveryTime":"v001","send_email":"pingtai@ichunt.com"}],"toUser":["pingtai@ichunt.com"],"ccUser":["zgj@ichunt.com","admin@ichunt.com","long@ichunt.com","zjl@ichunt.com"],"combined":"13.61","order_amount":"13.61","date":"2020-04-17","kefu_remark":"","id":"\u5f20\u5a77\u7389","order_id":3644,"operator_id":1000,"operator_event":"\u53d1\u9001\u90ae\u4ef6\u6210\u529f\uff0c\u6536\u4ef6\u4eba\u90ae\u7bb1\uff1a[\"pingtai@ichunt.com\"]\uff0c\u6284\u9001\u4eba\u90ae\u7bb1\uff1a[\"zgj@ichunt.com\",\"admin@ichunt.com\",\"long@ichunt.com\",\"zjl@ichunt.com\"]"}
+	 */
+
 	rabbitmq.Recv(rabbitmq.QueueExchange{
 		util.Configs.Rabbitmq_ichunt.QueueName,
 		util.Configs.Rabbitmq_ichunt.RoutingKey,

util/config.go

@@ -8,7 +8,10 @@ import (
 //订制配置文件解析载体
 type Config struct{
 	Liexin_databases *LiexinMysqlConfig
+	LiexinLabel_databases *LiexinLabelMysqlConfig
+	Liexincms_databases *LiexinCmsMysqlConfig
 	Rabbitmq_ichunt *RabbitmqIchunt
+	Rabbitmq_erp_label *RabbitmqErpLabel
 	Rabbitmq_order_push_stock *RabbitmqOrderPushStock
 	Crm_domain *SendMail
 	Ding_msg *Ding
@@ -21,6 +24,13 @@ type LiexinMysqlConfig struct{
 	Dns string `toml:"dns"`
 }
 
+type LiexinLabelMysqlConfig struct{
+	Dns string `toml:"dns"`
+}
+type LiexinCmsMysqlConfig struct{
+	Dns string `toml:"dns"`
+}
+
 type RabbitmqIchunt struct {
 	QueueName string `toml:"queue_name"`
 	RoutingKey string `toml:"routing_key"`
@@ -29,6 +39,14 @@ type RabbitmqIchunt struct {
 	Dns string `toml:"dns"`
 }
 
+type RabbitmqErpLabel struct {
+	QueueName string `toml:"queue_name"`
+	RoutingKey string `toml:"routing_key"`
+	Exchange string `toml:"exchange"`
+	Type string `toml:"type"`
+	Dns string `toml:"dns"`
+}
+
 type RabbitmqOrderPushStock struct {
 	QueueName string `toml:"queue_name"`
 	Dns string `toml:"dns"`