Merge branch 'dev' of http://119.23.72.7/sunlong_v5/kaopu-server into dev

cmd/dong/main.go

 package main
 package main
 
 
 import (
 import (
-	"flag"
 	"context"
 	"context"
-	"kaopu-server/internal/service"
+	"flag"
+	"fmt"
 	"github.com/bilibili/kratos/pkg/conf/paladin"
 	"github.com/bilibili/kratos/pkg/conf/paladin"
-	"kaopu-server/internal/logic"
 	"kaopu-server/internal/dao"
 	"kaopu-server/internal/dao"
-	"fmt"
+	"kaopu-server/internal/logic"
+	"kaopu-server/internal/service"
 	"os"
 	"os"
 	"strconv"
 	"strconv"
-	"sync"
+	"github.com/ichunt2019/go-msgserver/utils/rabbitmq"
+	"time"
 )
 )
 
 
+type RecvPro struct {}
+
 
 
 func main() {
 func main() {
 	//获取输入参数s
 	//获取输入参数s
@@ -37,29 +40,54 @@ func main() {
 		svc.DongRun(ctx,mod,"");  //运行main
 		svc.DongRun(ctx,mod,"");  //运行main
 	}else{
 	}else{
 
 
-		var wg sync.WaitGroup
-
 		for k,_ := range model_dong{
 		for k,_ := range model_dong{
-			// 开启协程
-			wg.Add(1)
-			go func(k int,ctx context.Context,mod_name string,mod_name_log string) {
 
 
-				fmt.Println("--开始协程:"+strconv.Itoa(k))
+			mod_name := dao.Model_dong[k]
+			mod_name_log :=dao.Model_dong_log[k]
+
+			fmt.Println("统计开始:"+strconv.Itoa(k))
 			logic.Loginfo("统计开始--"+mod_name,mod_name_log);
 			logic.Loginfo("统计开始--"+mod_name,mod_name_log);
 
 
 			svc := service.New()
 			svc := service.New()
 			svc.DongRun(ctx,mod,"");  //运行main
 			svc.DongRun(ctx,mod,"");  //运行main
 
 
-				fmt.Println("--结束协程:"+strconv.Itoa(k))
+			fmt.Println("统计结束:"+strconv.Itoa(k))
 			logic.Loginfo("统计结束--"+mod_name,mod_name_log);
 			logic.Loginfo("统计结束--"+mod_name,mod_name_log);
-
-				defer wg.Add(-1)
-			}(k,ctx,dao.Model_dong[k],dao.Model_dong_log[k])
 		}
 		}
 
 
-		wg.Wait()
 		fmt.Println("\r\n所有任务已经完成")
 		fmt.Println("\r\n所有任务已经完成")
 	}
 	}
 }
 }
 
 
+//队列监听
+func queues()  {
+	t := &RecvPro{}
+	queueExchange := &rabbitmq.QueueExchange{
+		"fengkong_dong_count",
+		"fengkong_dong_count",
+		"b_test_rabbit_mq",
+		"direct",
+		"amqp://guest:guest@192.168.2.232:5672/",
+	}
+
+	for{
+
+		mq := rabbitmq.New(queueExchange)
+		mq.RegisterReceiver(t)
+		err :=mq.Start()
+		if err != nil{
+
+			fmt.Println(err)
+		}
+		time.Sleep(time.Second)
+	}
+}
+
+//// 实现消费者 消费消息失败 自动进入延时尝试  尝试3次之后入库db
+func (t *RecvPro) Consumer(dataByte []byte) error {
+	
+	fmt.Println(string(dataByte))
+	//return errors.New("顶顶顶顶")
+	return nil
+}
 
 

internal/dao/dong.go

@@ -29,7 +29,7 @@ var Model_dong = map[int]string{
 	43:"退货率",
 	43:"退货率",
 	44:"近6个月逾期变化率",
 	44:"近6个月逾期变化率",
 	45:"近6个月参保人数变化率",
 	45:"近6个月参保人数变化率",
-	999:"统计动态分数",
+	999:"累计动态分数",
 }
 }
 //动态模型对应的日志文件名称
 //动态模型对应的日志文件名称
 var Model_dong_log = map[int]string{
 var Model_dong_log = map[int]string{
@@ -47,7 +47,6 @@ var Model_dong_log = map[int]string{
 
 
 //计算逾期金额率：延期支付时间超过平均账期的金额/收款总额:
 //计算逾期金额率：延期支付时间超过平均账期的金额/收款总额:
 func (d *dao) DongMain(ctx context.Context,project_id int,erp_company_code string) (r []map[string]interface{},err error) {
 func (d *dao) DongMain(ctx context.Context,project_id int,erp_company_code string) (r []map[string]interface{},err error) {
-
 	var (
 	var (
 		ladder_rate float64  //计算出来的各个动态分数率
 		ladder_rate float64  //计算出来的各个动态分数率
 		ladder_score float64 =0; //对应梯度分数
 		ladder_score float64 =0; //对应梯度分数
@@ -146,7 +145,10 @@ func (d *dao) DongMain(ctx context.Context,project_id int,erp_company_code strin
 			fmt.Print(log2)
 			fmt.Print(log2)
 			logic.Loginfo(log2,log_pre)
 			logic.Loginfo(log2,log_pre)
 		}
 		}
+		fmt.Print("XIE:"+v["erp_company_code"])
 	}
 	}
+
+	//d.Close(); //关闭数据库连接
 	logic.Loginfo("统计"+mod_name+"完成",log_pre)
 	logic.Loginfo("统计"+mod_name+"完成",log_pre)
 	return
 	return
 }
 }

internal/dao/scores.go

@@ -10,7 +10,7 @@ import (
 
 
 //增加或者修改记录
 //增加或者修改记录
 func (d *dao) InsertScores(ctx context.Context,score model.Scores) (insertId int64,err error) {
 func (d *dao) InsertScores(ctx context.Context,score model.Scores) (insertId int64,err error) {
-
+	//defer d.db.Close()
 	var(
 	var(
 		exitId int
 		exitId int
 		execSql string
 		execSql string
@@ -43,6 +43,6 @@ func (d *dao) InsertScores(ctx context.Context,score model.Scores) (insertId int
 	}
 	}
 	insertId = affectNum
 	insertId = affectNum
 
 
-	//defer d.db.Close()
+
 	return
 	return
 }
 }