up

go.mod

@@ -9,7 +9,7 @@ require (
 	github.com/go-sql-driver/mysql v1.4.1
 	github.com/gomodule/redigo v2.0.0+incompatible
 	github.com/ichunt2019/go-msgserver v1.0.4
-	github.com/ichunt2019/golang-rbmq-sl v0.0.0-20200515075131-59a37ab77d7d // indirect
+	github.com/ichunt2019/golang-rbmq-sl v0.0.0-20200515075131-59a37ab77d7d
 	github.com/ichunt2019/logger v1.0.5
 	github.com/jmoiron/sqlx v1.2.0
 	github.com/prometheus/common v0.7.0 // indirect

pyMonit/api-service-log/apiServiceLog.go

+package api_service_log
+
+import (
+	"encoding/json"
+	"fmt"
+	"github.com/ichunt2019/golang-rbmq-sl/utils/rabbitmq"
+	"time"
+)
+
+
+type DataParams struct {
+	InterfaceType string `json:"interface_type"`
+	AccessUrl string `json:"access_url"`
+	RequestParams string `json:"request_params"`
+	ErrMsg string `json:"err_msg"`
+	ErrCode string `json:"err_code"`
+	Uid string `json:"uid"`
+	UserName string `json:"user_name"`
+	UserIp string `json:"user_ip"`
+	Remakr string `json:"remark"`
+	CreateTime int64 `json:"create_time"`
+	CreateTimeStr string `json:"create_time_str"`
+}
+
+
+func AddMsgtoRbmq(msg string){
+
+	queueExchange := rabbitmq.QueueExchange{
+		"ichunt_monitor_user_behavior",
+		"ichunt_monitor_user_behavior",
+		"ichunt_monitor_behavior",
+		"direct",
+		"amqp://guest:guest@192.168.2.232:5672/",
+	}
+	
+	rbmqMsg := &DataParams{
+		InterfaceType: "20",
+		AccessUrl:     "",
+		RequestParams: "",
+		ErrMsg:        "python服务仓库推送监控超时",
+		ErrCode:       "",
+		Uid:           "",
+		UserName:      "",
+		UserIp:        "",
+		Remakr:        fmt.Sprintf("%s",msg),
+		CreateTime:    time.Now().Unix(),
+		CreateTimeStr: "",
+	}
+	_msg, _ := json.Marshal(rbmqMsg)
+	fmt.Println(string(_msg))
+
+	rabbitmq.Send(queueExchange,string(_msg))
+}
\ No newline at end of file

pyMonit/main.go

+package main
+
+import (
+	"flag"
+	"fmt"
+	"github.com/ichunt2019/golang-rbmq-sl/utils/rabbitmq"
+	"go-queue-server/dal/db"
+	api_service_log "go-queue-server/pyMonit/api-service-log"
+	_ "sync"
+	_ "time"
+)
+
+
+
+
+
+type RecvPro struct {
+
+}
+
+
+
+//// 实现消费者 消费消息失败 自动进入延时尝试  尝试3次之后入库db
+func (t *RecvPro) Consumer(dataByte []byte) error {
+	fmt.Println(string(dataByte))
+	api_service_log.AddMsgtoRbmq(string(dataByte))
+	return nil
+}
+
+//消息已经消费3次 失败了 请进行处理
+func (t *RecvPro) FailAction(err error,dataByte []byte) error {
+	//logger.Error("任务处理失败了，我要进入db日志库了")
+	//logger.Error("任务处理失败了，发送钉钉消息通知主人")
+	//logger.Error(string(dataByte))
+	//logger.Error("错误原因：%s",err)
+	//ding.Send(
+	//	util.Configs.Ding_msg.ErpLabelWebhook,
+	//	"@所有人 标签库系统告警:\n",
+	//	fmt.Sprintf("失败原因:%s \n %s",err,string(dataByte)) ,
+	//	make([]string,0),
+	//	false,
+	//	)
+	return nil
+}
+
+
+func initDb(action string,dns string) (err error) {
+	err = db.CreateDB(action,dns)
+	if err != nil {
+		fmt.Println(err)
+		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() {
+
+
+
+	t := &RecvPro{}
+
+	rabbitmq.Recv(rabbitmq.QueueExchange{
+		"api_service_log",
+		"api_service_log",
+		"",
+		"",
+		"amqp://guest:guest@192.168.2.232:5672/",
+	},t,1)
+}
\ No newline at end of file