获取任务接口, 改为长轮询方式

Author: ouqiang

README.md

@@ -15,7 +15,7 @@
 
 
 ## 运行
-`./delay-queue -c delay-queue.conf`, 默认监听 `0.0.0.0:9277`
+`./delay-queue -c delay-queue.conf`
 
 ## HTTP接口
 
@@ -56,7 +56,8 @@ URL地址 `/push`
 |   ttr  | int        |    Job执行超时时间, 单位：秒   |                   |
 |   body   | string     |    Job的内容，供消费者做具体的业务处理，如果是json格式需转义 |                   |
 
-### 获取ready queue中的任务    
+### 长轮询ready queue是否有任务, 有任务立即返回或等待180秒后超时返回   
+URL地址 `/pop`    
 ```json
 {
   "topic": "order"

config/config.go

@@ -17,11 +17,11 @@ const (
     DefaultRedisHost = "127.0.0.1:6379"
     DefaultRedisDb = 1
     DefaultRedisPassword = ""
-    DefaultRedisMaxIdle = 30
+    DefaultRedisMaxIdle = 10
     DefaultRedisMaxActive = 0
     DefaultRedisConnectTimeout = 5
-    DefaultRedisReadTimeout = 1
-    DefaultRedisWriteTimeout = 1
+    DefaultRedisReadTimeout = 3
+    DefaultRedisWriteTimeout = 3
 )
 
 type Config struct {

delayqueue/redis.go

@@ -63,6 +63,9 @@ func redisTestOnBorrow(conn redis.Conn, t time.Time) error {
         return nil
     }
     _, err := conn.Do("PING")
+    if err != nil {
+        log.Printf("从redis连接池取出的连接无效#%s", err.Error())
+    }
 
     return err
 }

routers/routers.go

@@ -72,12 +72,21 @@ func Pop(resp http.ResponseWriter, req *http.Request)  {
         resp.Write(generateFailureBody("topic不能为空"))
         return
     }
-    job, err := delayqueue.Pop(topic)
-    if err != nil {
-        log.Printf("获取job失败#%s", err.Error())
-        resp.Write(generateFailureBody("获取失败"))
-        return
+
+    // 轮询队列是否有任务, 有任务立即返回或180秒后超时返回
+    var job *delayqueue.Job = nil
+    waitTimeout := time.Now().Unix() + 180
+    for time.Now().Unix() < waitTimeout {
+        job, err = delayqueue.Pop(topic)
+        if job != nil && err == nil {
+            break
+        }
+        if err != nil {
+            log.Printf("获取job失败#%s", err.Error())
+        }
+        time.Sleep(1 * time.Second)
     }
+
     if job == nil {
         resp.Write(generateSuccessBody("操作成功", nil))
         return