up

app/service/http/service.go

@@ -31,6 +31,8 @@ type request struct {
 	TaskId int64 `json:"task_id"`
 	Header string `json:"header"`
 	IsHttpJson bool `json:"is_http_json"`
+	IsDelay bool `json:"is_delay"`
+	DelayTime int64 `json:"delay_time"`
 	Remark string `json:"remark"`
 	CallbackFunc string `json:"callback_func"`
 }
@@ -44,7 +46,20 @@ func addMsg(msg string) (err error){
 		lib.Instance("config").GetString("rabbitmq_golang_async_task.type"),
 		lib.Instance("config").GetString("rabbitmq_golang_async_task.dns"),
 	}
-	rabbitmq.Send(queueExchange,msg)
+	_ = rabbitmq.Send(queueExchange,msg)
+	return
+}
+
+//延时消息
+func addDelayMsg(msg string,ttl int64) (err error){
+	queueExchange := rabbitmq.QueueExchange{
+		lib.Instance("config").GetString("rabbitmq_golang_async_task.queue_name"),
+		lib.Instance("config").GetString("rabbitmq_golang_async_task.routing_key"),
+		lib.Instance("config").GetString("rabbitmq_golang_async_task.exchange"),
+		lib.Instance("config").GetString("rabbitmq_golang_async_task.type"),
+		lib.Instance("config").GetString("rabbitmq_golang_async_task.dns"),
+	}
+	_ = rabbitmq.SendDelay(queueExchange,msg,ttl)
 	return
 }
 
@@ -59,13 +74,22 @@ func Callback(ctx *gin.Context) (err error){
 	req.ServiceType,_   = strconv.Atoi(serviceType)
 	req.Header         = ctx.DefaultPostForm("header","")
 	is_http_json       := ctx.DefaultPostForm("isHttpJson","0")
+	is_delay            := ctx.DefaultPostForm("is_delay","0")
+	delay_time_str      := ctx.DefaultPostForm("delay_time","0")
+
 	req.IsHttpJson,_    = strconv.ParseBool(is_http_json)
+	req.IsDelay,_    = strconv.ParseBool(is_delay)
 	req.Remark          = ctx.DefaultPostForm("remark","")
 	req.CallbackFunc    = ctx.DefaultPostForm("callback_func","")
 
 	req.RequestType     = strings.ToLower(req.RequestType)
 	req.Method          = strings.ToLower(req.Method)
 
+	delay_time, err := strconv.ParseInt(delay_time_str, 10, 64)
+	if err == nil{
+		req.DelayTime = delay_time
+	}
+
 	//字段验证
 	if !php2go.InArray(req.RequestType,[]string{"http","hprose"}){
 		return errors.New("The field requestType must be in  http or hprose, The default is http")
@@ -115,7 +139,12 @@ func Callback(ctx *gin.Context) (err error){
 
 	if msg ,err :=json.Marshal(req); err == nil{
 		//插入队列
-		_ = addMsg(string(msg))
+		if req.IsDelay &&  req.DelayTime > 0{
+			//延时
+			_ = addDelayMsg(string(msg),req.DelayTime)
+		}else{
+			_ = addMsg(string(msg))
+		}
 		err = session.Commit()
 	}else{
 		err = session.Rollback()

go.mod

@@ -24,7 +24,7 @@ require (
 	github.com/hprose/hprose-golang v2.0.6+incompatible
 	github.com/ichunt2019/cfg v0.0.0-20210310074903-4b1bcab17717
 	github.com/ichunt2019/go-redis-pool v0.0.0-20210305064829-86b9011c57f5
-	github.com/ichunt2019/golang-rbmq-sl v0.0.0-20200515075131-59a37ab77d7d
+	github.com/ichunt2019/golang-rbmq-sl v0.0.0-20211021112417-f04f7df9324a
 	github.com/ichunt2019/ichunt-micro-registry v1.0.2
 	github.com/ichunt2019/lxLog v0.0.0-20210526032824-fd07e8b73ce0
 	github.com/lib/pq v1.9.0 // indirect

go.sum

@@ -244,6 +244,10 @@ github.com/ichunt2019/go-redis-pool v0.0.0-20210305064829-86b9011c57f5 h1:oP7r2x
 github.com/ichunt2019/go-redis-pool v0.0.0-20210305064829-86b9011c57f5/go.mod h1:63lgYPBXcNBPUtKz14iQeFZBm2318xG/zGvCczRlfSs=
 github.com/ichunt2019/golang-rbmq-sl v0.0.0-20200515075131-59a37ab77d7d h1:wN4ay65hYnXyqn1BWc/WgCpPf+IYwYDRbLASYZyQrPw=
 github.com/ichunt2019/golang-rbmq-sl v0.0.0-20200515075131-59a37ab77d7d/go.mod h1:sUQJFISXhgUd5EjkbKphSoxHiGL5BMkTQ/9rfT0lVCw=
+github.com/ichunt2019/golang-rbmq-sl v0.0.0-20211021095324-74d9097e847a h1:DkB7uXqYOO849R/3jTQ5o2ZsE6fp8/0Vj0KN8hjV0Lg=
+github.com/ichunt2019/golang-rbmq-sl v0.0.0-20211021095324-74d9097e847a/go.mod h1:sUQJFISXhgUd5EjkbKphSoxHiGL5BMkTQ/9rfT0lVCw=
+github.com/ichunt2019/golang-rbmq-sl v0.0.0-20211021112417-f04f7df9324a h1:Qhqadsbh5D19B55ZTXuEsXbUbHGWUDAv0wmLmPDGxoo=
+github.com/ichunt2019/golang-rbmq-sl v0.0.0-20211021112417-f04f7df9324a/go.mod h1:sUQJFISXhgUd5EjkbKphSoxHiGL5BMkTQ/9rfT0lVCw=
 github.com/ichunt2019/ichunt-micro-registry v1.0.1 h1:IrNy60FRIPzqf+Fk/hH5BbEwQgxfzpz62nNNTqpl2x8=
 github.com/ichunt2019/ichunt-micro-registry v1.0.1/go.mod h1:OPGbBtYiHTjmxOZ43l32PLsiw+89CEDANx/Cs5Y9zkQ=
 github.com/ichunt2019/ichunt-micro-registry v1.0.2 h1:hZ68McTgNvfsHjxlU8QgoHLF8UCAu0buju6KI1qtpIU=