1.0

cmd/dong/main.go

@@ -10,6 +10,7 @@ import (
 	"kaopu-server/internal/service"
 	"kaopu-server/internal/service"
 	"os"
 	"os"
 	"strconv"
 	"strconv"
+	"errors"
 	"github.com/ichunt2019/go-msgserver/utils/rabbitmq"
 	"github.com/ichunt2019/go-msgserver/utils/rabbitmq"
 	"time"
 	"time"
 )
 )
@@ -25,6 +26,10 @@ func main() {
 	if err := paladin.Init(); err != nil {
 	if err := paladin.Init(); err != nil {
 		panic(err)
 		panic(err)
 	}
 	}
+	if mod == -1 {
+		queues() //队列监听
+	}
+
 	model_dong := dao.Model_dong //所有动态模型算法列表
 	model_dong := dao.Model_dong //所有动态模型算法列表
 
 
 	_ ,ok := model_dong[mod]
 	_ ,ok := model_dong[mod]
@@ -35,13 +40,18 @@ func main() {
 
 
 	ctx := context.Background()  //上下文
 	ctx := context.Background()  //上下文
 
 
+	if mod == -1 {
+		queues() //队列监听
+	}
 	if mod >0 {
 	if mod >0 {
 		svc := service.New()
 		svc := service.New()
 		svc.DongRun(ctx,mod,"");  //运行main
 		svc.DongRun(ctx,mod,"");  //运行main
 	}else{
 	}else{
 
 
 		for k,_ := range model_dong{
 		for k,_ := range model_dong{
-
+			if k == -1 || k == 0 {
+				continue;
+			}
 			mod_name := dao.Model_dong[k]
 			mod_name := dao.Model_dong[k]
 			mod_name_log :=dao.Model_dong_log[k]
 			mod_name_log :=dao.Model_dong_log[k]
 
 
@@ -62,14 +72,14 @@ func main() {
 //队列监听
 //队列监听
 func queues()  {
 func queues()  {
 	t := &RecvPro{}
 	t := &RecvPro{}
+
 	queueExchange := &rabbitmq.QueueExchange{
 	queueExchange := &rabbitmq.QueueExchange{
 		"fengkong_dong_count",
 		"fengkong_dong_count",
 		"fengkong_dong_count",
 		"fengkong_dong_count",
-		"b_test_rabbit_mq",
+		"fengkong_exchange",
 		"direct",
 		"direct",
 		"amqp://guest:guest@192.168.2.232:5672/",
 		"amqp://guest:guest@192.168.2.232:5672/",
 	}
 	}
-
 	for{
 	for{
 
 
 		mq := rabbitmq.New(queueExchange)
 		mq := rabbitmq.New(queueExchange)
@@ -81,13 +91,20 @@ func queues()  {
 		}
 		}
 		time.Sleep(time.Second)
 		time.Sleep(time.Second)
 	}
 	}
+
 }
 }
 
 
 //// 实现消费者 消费消息失败 自动进入延时尝试  尝试3次之后入库db
 //// 实现消费者 消费消息失败 自动进入延时尝试  尝试3次之后入库db
 func (t *RecvPro) Consumer(dataByte []byte) error {
 func (t *RecvPro) Consumer(dataByte []byte) error {
 
 
 	fmt.Println(string(dataByte))
 	fmt.Println(string(dataByte))
-	//return errors.New("顶顶顶顶")
+	return errors.New("顶顶顶顶")
+	//return nil
+}
+//消息已经消费3次 失败了 请进行处理
+func (t *RecvPro) FailAction(dataByte []byte) error {
+	fmt.Println(string(dataByte))
+	fmt.Println("任务处理失败了，我要进入db日志库了")
+	fmt.Println("任务处理失败了，发送钉钉消息通知主人")
 	return nil
 	return nil
 }
 }
-

go.mod

@@ -4,7 +4,7 @@ go 1.12
 
 
 require (
 require (
 	github.com/bilibili/kratos v0.3.1
 	github.com/bilibili/kratos v0.3.1
-	github.com/ichunt2019/go-msgserver v0.0.0-20191204053231-bd327f4f8ff1
+	github.com/ichunt2019/go-msgserver v0.0.0-20191205082056-967d5dbbe7cd
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/pkg/errors v0.8.1
 	github.com/pkg/errors v0.8.1
 )
 )

internal/dao/dong.go

@@ -30,6 +30,7 @@ var Model_dong = map[int]string{
 	44:"近6个月逾期变化率",
 	44:"近6个月逾期变化率",
 	45:"近6个月参保人数变化率",
 	45:"近6个月参保人数变化率",
 	999:"累计动态分数",
 	999:"累计动态分数",
+	-1:"队列监听",
 }
 }
 //动态模型对应的日志文件名称
 //动态模型对应的日志文件名称
 var Model_dong_log = map[int]string{
 var Model_dong_log = map[int]string{