feat: 消费者轮询队列, 支持单个连接同时轮询多个队列

Author: ouqiang

delayqueue/delay_queue.go

@@ -42,8 +42,8 @@ func Push(job Job) error {
 }
 
 // 获取Job
-func Pop(topic string) (*Job, error) {
-     jobId, err := blockPopFromReadyQueue(topic, config.Setting.QueueBlockTimeout)
+func Pop(topics []string) (*Job, error) {
+     jobId, err := blockPopFromReadyQueue(topics, config.Setting.QueueBlockTimeout)
      if err != nil {
          return nil, err
      }

delayqueue/ready_queue.go

@@ -14,9 +14,14 @@ func pushToReadyQueue(queueName string, jobId string) error {
     return err
 }
 
-func blockPopFromReadyQueue(queueName string, timeout int) (string, error){
-    queueName = fmt.Sprintf(config.Setting.QueueName, queueName)
-    value, err := execRedisCommand("BLPOP", queueName, timeout)
+func blockPopFromReadyQueue(queues []string, timeout int) (string, error){
+    var args []interface{}
+    for _, queue := range queues {
+        queue = fmt.Sprintf(config.Setting.QueueName, queue)
+        args = append(args, queue)
+    }
+    args = append(args, timeout)
+    value, err := execRedisCommand("BLPOP", args...)
     if err != nil {
         return "", err
     }

routers/routers.go

@@ -72,8 +72,9 @@ func Pop(resp http.ResponseWriter, req *http.Request)  {
         resp.Write(generateFailureBody("topic不能为空"))
         return
     }
-
-    job, err := delayqueue.Pop(topic)
+    // 多个topic逗号分隔
+    topics := strings.Split(topic, ",")
+    job, err := delayqueue.Pop(topics)
     if err != nil {
         log.Printf("获取job失败#%s", err.Error())
         resp.Write(generateFailureBody("获取Job失败"))