refactor(获取Job的方式由轮询改为调用blpop阻塞获取):

ouqiang · ouqiang · commit d02bd9cad65e · 2017-07-03T17:32:20.000+08:00
diff --git a/README.md b/README.md
@@ -13,7 +13,9 @@
 
 
 ## 运行
-`./delay-queue -c delay-queue.conf`  默认监听`0.0.0.0:9277`   
+`./delay-queue -c delay-queue.conf`  
+> HTTP Server监听`0.0.0.0:9277`, Redis连接地址`127.0.0.1:6379`, 数据库编号`1`
+   
 
 ## HTTP接口
 
diff --git a/config/config.go b/config/config.go
@@ -13,21 +13,27 @@ var (
 
 const (
     DefaultBindAddress = "0.0.0.0:9277"
-    DefaultBucket = 5;
+    DefaultBucketSize = 3
+    DefaultBucketName = "dq_bucket_%d"
+    DefaultQueueName = "dq_queue_%s"
+    DefaultQueueBlockTimeout = 178
     DefaultRedisHost = "127.0.0.1:6379"
     DefaultRedisDb = 1
     DefaultRedisPassword = ""
     DefaultRedisMaxIdle = 10
     DefaultRedisMaxActive = 0
-    DefaultRedisConnectTimeout = 5
-    DefaultRedisReadTimeout = 3
-    DefaultRedisWriteTimeout = 3
+    DefaultRedisConnectTimeout = 5000
+    DefaultRedisReadTimeout = 180000
+    DefaultRedisWriteTimeout = 3000
 )
 
 type Config struct {
-    BindAddress string
-    Bucket int
-    Redis RedisConfig
+    BindAddress string    // http server 监听地址
+    BucketSize int        // bucket数量
+    BucketName string     // bucket在redis中的键名,
+    QueueName string      // ready queue在redis中的键名
+    QueueBlockTimeout int // 调用blpop阻塞超时时间, 单位秒, 修改此项, redis.read_timeout必须做相应调整
+    Redis RedisConfig     // redis配置
 }
 
 type RedisConfig struct {
@@ -36,9 +42,9 @@ type RedisConfig struct {
     Password string
     MaxIdle int    // 连接池最大空闲连接数
     MaxActive int  // 连接池最大激活连接数
-    ConnectTimeout int  // 连接超时, 单位秒
-    ReadTimeout int     // 读取超时, 单位秒
-    WriteTimeout int    // 写入超时, 单位秒
+    ConnectTimeout int  // 连接超时, 单位毫秒
+    ReadTimeout int     // 读取超时, 单位毫秒
+    WriteTimeout int    // 写入超时, 单位毫秒
 }
 
 func Init(path string)  {
@@ -59,7 +65,11 @@ func (config *Config) parse(path string)  {
 
     section := file.Section("")
     config.BindAddress = section.Key("bind_address").MustString(DefaultBindAddress)
-    config.Bucket = section.Key("bucket").MustInt(DefaultBucket)
+    config.BucketSize = section.Key("bucket_size").MustInt(DefaultBucketSize)
+    config.BucketName = section.Key("bucket_name").MustString(DefaultBucketName)
+    config.QueueName = section.Key("queue_name").MustString(DefaultQueueName)
+    config.QueueBlockTimeout = section.Key("queue_block_timeout").MustInt(DefaultQueueBlockTimeout)
+
     config.Redis.Host = section.Key("redis.host").MustString(DefaultRedisHost)
     config.Redis.Db = section.Key("redis.db").MustInt(DefaultRedisDb)
     config.Redis.Password = section.Key("redis.password").MustString(DefaultRedisPassword)
@@ -73,7 +83,11 @@ func (config *Config) parse(path string)  {
 
 func (config *Config) initDefaultConfig()  {
     config.BindAddress = DefaultBindAddress
-    config.Bucket = DefaultBucket
+    config.BucketSize = DefaultBucketSize
+    config.BucketName = DefaultBucketName
+    config.QueueName = DefaultQueueName
+    config.QueueBlockTimeout = DefaultQueueBlockTimeout
+
     config.Redis.Host = DefaultRedisHost
     config.Redis.Db = DefaultRedisDb
     config.Redis.Password = DefaultRedisPassword
diff --git a/delay-queue.conf b/delay-queue.conf
@@ -1,10 +1,13 @@
-
-# http server监听地址
-bind_address = '0.0.0.0:9277'
-# bucket数量　
-bucket = 5
-
-redis.host = 127.0.0.1:6379
-redis.db = 1
-# 连接不需要密码留空
-redis.password = ""
+;bind_address = 0.0.0.0:9278   ; http server监听地址
+;bucket_size = 3               ; bucket数量
+;bucket_name = dq_bucket_%d    ; bucket在redis中的键名, %d必须保留
+;queue_name = dq_queue_%s      ; ready queue在redis中的键名, %s必须保留
+;queue_block_timeout = 178     ; 调用blpop阻塞超时时间, 单位秒, 必须小于redis.read_timeout, 修改此项, redis.read_timeout需做相应调整
+;redis.host = 127.0.0.1:6379   ; redis连接地址
+;redis.db = 1                  ; 建议选择未被使用的数据库或新建redis实例，防止key冲突
+;redis.password =              ; redis密码, 无需密码留空
+;redis.max_idle = 10           ; redis连接池最大空闲连接数
+;redis.max_active = 0          ; redis连接池最大激活连接数, 0为不限制
+;redis.connect_timeout = 5000  ; redis连接超时时间, 单位毫秒
+;redis.read_timeout = 180000   ; redis读取超时时间, 单位毫秒
+;redis.write_timeout = 3000    ; redis写入超时时间, 单位毫秒
diff --git a/delayqueue/delay_queue.go b/delayqueue/delay_queue.go
@@ -7,15 +7,12 @@ import (
     "log"
     "errors"
 )
-const (
-    BucketRedisKey = "dq_bucket_%d"
-)
 
 var (
     // 每个定时器对应一个bucket
     timers []*time.Ticker
     // bucket名称chan
-    bucketNameChan chan string
+    bucketNameChan <- chan string
 )
 
 func Init()  {
@@ -46,7 +43,7 @@ func Push(job Job) error {
 
 // 获取Job
 func Pop(topic string) (*Job, error) {
-     jobId, err := popFromReadyQueue(topic)
+     jobId, err := blockPopFromReadyQueue(topic, config.Setting.QueueBlockTimeout)
      if err != nil {
          return nil, err
      }
@@ -79,13 +76,13 @@ func Remove(jobId string) error {
 }
 
 // 轮询获取Job名称, 使job分布到不同bucket中, 提高扫描速度
-func generateBucketName() (chan string) {
+func generateBucketName() (<- chan string) {
     c := make(chan string)
     go func() {
         i := 1
         for {
-            c <- fmt.Sprintf(BucketRedisKey, i)
-            if i >= config.Setting.Bucket {
+            c <- fmt.Sprintf(config.Setting.BucketName, i)
+            if i >= config.Setting.BucketSize {
                 i = 1
             } else {
                 i++
@@ -98,11 +95,11 @@ func generateBucketName() (chan string) {
 
 // 初始化定时器
 func initTimers()  {
-    timers = make([]*time.Ticker, config.Setting.Bucket)
+    timers = make([]*time.Ticker, config.Setting.BucketSize)
     var bucketName string
-    for i := 0; i < config.Setting.Bucket; i++ {
+    for i := 0; i < config.Setting.BucketSize; i++ {
         timers[i] = time.NewTicker(1 * time.Second)
-        bucketName = fmt.Sprintf(BucketRedisKey, i + 1)
+        bucketName = fmt.Sprintf(config.Setting.BucketName, i + 1)
         go waitTicker(timers[i], bucketName)
     }
 }
diff --git a/delayqueue/ready_queue.go b/delayqueue/ready_queue.go
@@ -1,23 +1,34 @@
 package delayqueue
 
+import (
+    "fmt"
+    "github.com/ouqiang/delay-queue/config"
+)
 
 type ReadyQueue struct {}
 
 func pushToReadyQueue(queueName string, jobId string) error {
+    queueName = fmt.Sprintf(config.Setting.QueueName, queueName)
     _, err := execRedisCommand("RPUSH", queueName, jobId)
 
     return err
 }
 
-func popFromReadyQueue(queueName string) (string, error){
-    value, err := execRedisCommand("LPOP", queueName)
+func blockPopFromReadyQueue(queueName string, timeout int) (string, error){
+    queueName = fmt.Sprintf(config.Setting.QueueName, queueName)
+    value, err := execRedisCommand("BLPOP", queueName, timeout)
     if err != nil {
         return "", err
     }
     if value == nil {
         return "", nil
     }
-    byteValue := value.([]byte)
+    var valueBytes []interface{}
+    valueBytes = value.([]interface{})
+    if len(valueBytes) == 0 {
+        return "", nil
+    }
+    element := string(valueBytes[1].([]byte))
 
-    return string(byteValue), nil
+    return element, nil
 }
diff --git a/delayqueue/redis.go b/delayqueue/redis.go
@@ -30,9 +30,9 @@ func redisDial() (redis.Conn, error)  {
     conn, err := redis.Dial(
             "tcp",
             config.Setting.Redis.Host,
-            redis.DialConnectTimeout(time.Duration(config.Setting.Redis.ConnectTimeout) * time.Second),
-            redis.DialReadTimeout(time.Duration(config.Setting.Redis.ReadTimeout) * time.Second),
-            redis.DialWriteTimeout(time.Duration(config.Setting.Redis.WriteTimeout) * time.Second),
+            redis.DialConnectTimeout(time.Duration(config.Setting.Redis.ConnectTimeout) * time.Millisecond),
+            redis.DialReadTimeout(time.Duration(config.Setting.Redis.ReadTimeout) * time.Millisecond),
+            redis.DialWriteTimeout(time.Duration(config.Setting.Redis.WriteTimeout) * time.Millisecond),
     )
     if err != nil {
         log.Printf("连接redis失败#%s", err.Error())
diff --git a/routers/routers.go b/routers/routers.go
@@ -73,18 +73,11 @@ func Pop(resp http.ResponseWriter, req *http.Request)  {
         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)
+    job, err := delayqueue.Pop(topic)
+    if err != nil {
+        log.Printf("获取job失败#%s", err.Error())
+        resp.Write(generateFailureBody("获取Job失败"))
+        return
     }
 
     if job == nil {
@@ -102,7 +95,7 @@ func Pop(resp http.ResponseWriter, req *http.Request)  {
         Body: job.Body,
     }
 
-    log.Printf("获取job#%+v", data)
+    log.Printf("get job#%+v", data)
 
     resp.Write(generateSuccessBody("操作成功", data))
 }
