feat: add send message async

main.go

@@ -7,7 +7,9 @@ import (
 	"message-nest/pkg/util"
 	"message-nest/service/cron_service"
 	"message-nest/service/env_service"
+	"message-nest/service/send_message_service"
 	"net/http"
+	"sync"
 
 	"github.com/gin-gonic/gin"
 
@@ -20,6 +22,8 @@ import (
 var (
 	//go:embed web/dist/*
 	f embed.FS
+
+	wg sync.WaitGroup
 )
 
 func init() {
@@ -32,7 +36,7 @@ func init() {
 	cron_service.Setup()
 }
 
-func main() {
+func GinServerUp() {
 	gin.SetMode(setting.ServerSetting.RunMode)
 
 	routersInit := routers.InitRouter(f)
@@ -56,3 +60,13 @@ func main() {
 		logging.Logger.Error("Server err: ", err)
 	}
 }
+
+func main() {
+	wg.Add(1)
+
+	go GinServerUp()
+	go send_message_service.MessageConsumer(&wg)
+
+	wg.Wait()
+	fmt.Println("Server exit...")
+}

pkg/constant/constant.go

@@ -26,3 +26,6 @@ var LogsCleanDefaultValueMap = map[string]string{
 }
 
 const AboutSectionName = "about"
+
+// 限制goroutine的最大数量
+var MaxSemaphore = make(chan struct{}, 2048)

routers/api/v1/send_message.go

@@ -13,6 +13,7 @@ type SendMessageReq struct {
 	Text     string `json:"text" validate:"required" label:"文本内容"`
 	HTML     string `json:"html"  label:"html内容"`
 	MarkDown string `json:"markdown" label:"markdown内容"`
+	Mode     string `json:"mode" label:"是否异步发送"`
 }
 
 // DoSendMassage 外部调用发信接口
@@ -34,11 +35,17 @@ func DoSendMassage(c *gin.Context) {
 		HTML:     req.HTML,
 		MarkDown: req.MarkDown,
 	}
-	err := msgService.Send()
-	if err != "" {
-		appG.CResponse(http.StatusBadRequest, "发送失败！", nil)
-		return
+	if req.Mode != "async" {
+		// 同步发送
+		err := msgService.Send()
+		if err != "" {
+			appG.CResponse(http.StatusBadRequest, "发送失败！", nil)
+			return
+		}
+		appG.CResponse(http.StatusOK, "发送成功！", nil)
+	} else {
+		// 异步发送
+		send_message_service.Buffer <- msgService
+		appG.CResponse(http.StatusOK, "提交成功！", nil)
 	}
-
-	appG.CResponse(http.StatusOK, "发送成功！", nil)
 }

service/send_message_service/consumer.go

@@ -0,0 +1,42 @@
+package send_message_service
+
+import (
+	"message-nest/pkg/constant"
+	"message-nest/pkg/logging"
+	"sync"
+)
+
+var maxBufferSize = 10
+var Buffer = make(chan SendMessageService, maxBufferSize)
+
+func DoSendTask(task SendMessageService, wg *sync.WaitGroup) {
+	defer wg.Done()
+	defer func() {
+		if r := recover(); r != nil {
+			logging.Logger.Error("DoSendTask: Recovered from panic:", r)
+		}
+	}()
+
+	constant.MaxSemaphore <- struct{}{}
+	defer func() {
+		<-constant.MaxSemaphore
+	}()
+
+	go task.Send()
+}
+
+func MessageConsumer(wg *sync.WaitGroup) {
+	defer wg.Done()
+
+	for {
+		task, ok := <-Buffer
+		if !ok {
+			logging.Logger.Error("MessageConsumer: Channel closed. Exiting.")
+			return
+		}
+
+		wg.Add(1)
+		go DoSendTask(task, wg)
+	}
+
+}

service/send_message_service/send_message.go

@@ -37,6 +37,7 @@ func (sm *SendMessageService) LogsAndStatusMark(errStr string, status int) {
 	if status == SendFail {
 		sm.Status = SendFail
 	}
+	logging.Logger.Info(fmt.Sprintf("%s, 状态：%d", errStr, status))
 }
 
 // Send 发送一个消息任务的所有实例

web/src/util/viewApi.js

@@ -8,7 +8,7 @@ class ApiStrGenerate {
 
 
     static getCurlString(task_id, options) {
-        let data = { task_id: task_id };
+        let data = { task_id: task_id, mode: "async" };
         data.text = 'Hello World!';
         if (options.html) {
             data.html = '<h1> Hello World! </h1>';

web/src/views/tabsTools/sendTasks/view/viewTaskAPIPopUp.vue

@@ -10,7 +10,7 @@
           <br />
           发送的消息会优先现在相应的类型消息进行发送，如果没有，将使用传的text消息进行发送          
           <br />
-          *text节点必传
+          ** text节点必传，指定mode=async将异步发送，否则同步发送
         </template>
         <el-icon>
           <QuestionFilled />