增加Redis连接、读取、写入超时配置

Author: ouqiang

cmd/cmd.go

@@ -9,6 +9,7 @@ import (
     "github.com/ouqiang/delay-queue/delayqueue"
     "log"
     "fmt"
+    _ "net/http/pprof"
 )
 
 type Cmd struct {}

config/config.go

@@ -19,6 +19,9 @@ const (
     DefaultRedisPassword = ""
     DefaultRedisMaxIdle = 30
     DefaultRedisMaxActive = 0
+    DefaultRedisConnectTimeout = 5
+    DefaultRedisReadTimeout = 1
+    DefaultRedisWriteTimeout = 1
 )
 
 type Config struct {
@@ -33,6 +36,9 @@ type RedisConfig struct {
     Password string
     MaxIdle int    // 连接池最大空闲连接数
     MaxActive int  // 连接池最大激活连接数
+    ConnectTimeout int  // 连接超时, 单位秒
+    ReadTimeout int     // 读取超时, 单位秒
+    WriteTimeout int    // 写入超时, 单位秒
 }
 
 func Init(path string)  {
@@ -59,6 +65,9 @@ func (config *Config) parse(path string)  {
     config.Redis.Password = section.Key("redis.password").MustString(DefaultRedisPassword)
     config.Redis.MaxIdle = section.Key("redis.max_idle").MustInt(DefaultRedisMaxIdle)
     config.Redis.MaxActive = section.Key("redis.max_active").MustInt(DefaultRedisMaxActive)
+    config.Redis.ConnectTimeout = section.Key("redis.connect_timeout").MustInt(DefaultRedisConnectTimeout)
+    config.Redis.ReadTimeout = section.Key("redis.read_timeout").MustInt(DefaultRedisReadTimeout)
+    config.Redis.WriteTimeout = section.Key("redis.write_timeout").MustInt(DefaultRedisWriteTimeout)
 }
 
 
@@ -70,4 +79,7 @@ func (config *Config) initDefaultConfig()  {
     config.Redis.Password = DefaultRedisPassword
     config.Redis.MaxIdle = DefaultRedisMaxIdle
     config.Redis.MaxActive = DefaultRedisMaxActive
+    config.Redis.ConnectTimeout = DefaultRedisConnectTimeout
+    config.Redis.ReadTimeout = DefaultRedisReadTimeout
+    config.Redis.WriteTimeout = DefaultRedisWriteTimeout
 }

delayqueue/delay_queue.go

@@ -5,6 +5,7 @@ import (
     "github.com/ouqiang/delay-queue/config"
     "fmt"
     "log"
+    "errors"
 )
 const (
     BucketRedisKey = "dq_bucket_%d"
@@ -13,27 +14,34 @@ const (
 var (
     // 每个定时器对应一个bucket
     timers []*time.Ticker
-    // 保存待放入bucket中的job
-    queue chan Job
     // bucket名称chan
     bucketNameChan chan string
 )
 
 func Init()  {
     RedisPool = initRedisPool()
     initTimers()
-    queue = make(chan Job, 100)
     bucketNameChan = generateBucketName()
-    go waitJob()
 }
 
 // 添加一个Job到队列中
-func Push(job Job)  {
+func Push(job Job) error {
     if job.Id == "" || job.Topic == "" || job.Delay < 0 || job.TTR <= 0 {
-        return
+        return errors.New("invalid job")
     }
 
-    queue <- job
+    err := putJob(job.Id, job)
+    if err != nil {
+        log.Printf("添加job到job pool失败#job-%+v#%s", job, err.Error())
+        return err
+    }
+    err = pushToBucket(<-bucketNameChan, job.Delay, job.Id)
+    if err != nil {
+        log.Printf("添加job到bucket失败#job-%+v#%s", job, err.Error())
+        return err
+    }
+
+    return nil
 }
 
 // 获取Job
@@ -70,22 +78,6 @@ func Remove(jobId string) error {
     return removeJob(jobId)
 }
 
-
-func waitJob()  {
-    var err error
-    for job := range queue {
-        err = putJob(job.Id, job)
-        if err != nil {
-            log.Printf("添加job到job pool失败#job-%+v#%s", job, err.Error())
-            continue
-        }
-        err = pushToBucket(<-bucketNameChan, job.Delay, job.Id)
-        if err != nil {
-            log.Printf("添加job到bucket失败#job-%+v#%s", job, err.Error())
-        }
-    }
-}
-
 // 轮询获取Job名称, 使job分布到不同bucket中, 提高扫描速度
 func generateBucketName() (chan string) {
     c := make(chan string)

delayqueue/redis.go

@@ -27,7 +27,13 @@ func initRedisPool() *redis.Pool {
 
 // 连接redis
 func redisDial() (redis.Conn, error)  {
-    conn, err := redis.Dial("tcp", config.Setting.Redis.Host)
+    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),
+    )
     if err != nil {
         log.Printf("连接redis失败#%s", err.Error())
         return nil, err
@@ -43,10 +49,12 @@ func redisDial() (redis.Conn, error)  {
 
     _, err = conn.Do("SELECT", config.Setting.Redis.Db)
     if err != nil {
+        conn.Close()
         log.Printf("redis选择数据库失败#%s", err.Error())
+        return nil, err
     }
 
-    return conn, err
+    return conn, nil
 }
 
 // 从池中取出连接后，判断连接是否有效

routers/routers.go

@@ -51,9 +51,13 @@ func Push(resp http.ResponseWriter, req *http.Request)  {
 
     log.Printf("add job#%+v\n", job)
     job.Delay = time.Now().Unix() + job.Delay
-    delayqueue.Push(job)
+    err = delayqueue.Push(job)
 
-    resp.Write(generateSuccessBody("添加成功", nil))
+    if err != nil {
+        resp.Write(generateFailureBody("添加失败"))
+    } else {
+        resp.Write(generateSuccessBody("添加成功", nil))
+    }
 }
 
 // 获取job