Merge pull request #460 from pengfei-xue/develop
use connection pool for redis, support auto connection
Showing
1 changed file
with
34 additions
and
112 deletions
| ... | @@ -3,7 +3,7 @@ package cache | ... | @@ -3,7 +3,7 @@ package cache |
| 3 | import ( | 3 | import ( |
| 4 | "encoding/json" | 4 | "encoding/json" |
| 5 | "errors" | 5 | "errors" |
| 6 | "io" | 6 | "time" |
| 7 | 7 | ||
| 8 | "github.com/beego/redigo/redis" | 8 | "github.com/beego/redigo/redis" |
| 9 | ) | 9 | ) |
| ... | @@ -15,7 +15,7 @@ var ( | ... | @@ -15,7 +15,7 @@ var ( |
| 15 | 15 | ||
| 16 | // Redis cache adapter. | 16 | // Redis cache adapter. |
| 17 | type RedisCache struct { | 17 | type RedisCache struct { |
| 18 | c redis.Conn | 18 | p *redis.Pool // redis connection pool |
| 19 | conninfo string | 19 | conninfo string |
| 20 | key string | 20 | key string |
| 21 | } | 21 | } |
| ... | @@ -25,23 +25,17 @@ func NewRedisCache() *RedisCache { | ... | @@ -25,23 +25,17 @@ func NewRedisCache() *RedisCache { |
| 25 | return &RedisCache{key: DefaultKey} | 25 | return &RedisCache{key: DefaultKey} |
| 26 | } | 26 | } |
| 27 | 27 | ||
| 28 | // Get cache from redis. | 28 | // actually do the redis cmds |
| 29 | func (rc *RedisCache) Get(key string) interface{} { | 29 | func (rc *RedisCache) do(commandName string, args ...interface{}) (reply interface{}, err error) { |
| 30 | if rc.c == nil { | 30 | c := rc.p.Get() |
| 31 | var err error | 31 | defer c.Close() |
| 32 | rc.c, err = rc.connectInit() | ||
| 33 | if err != nil { | ||
| 34 | return nil | ||
| 35 | } | ||
| 36 | } | ||
| 37 | 32 | ||
| 38 | v, err := rc.c.Do("HGET", rc.key, key) | 33 | return c.Do(commandName, args...) |
| 39 | // write to closed socket, reset rc.c to nil | 34 | } |
| 40 | if err == io.EOF { | ||
| 41 | rc.c = nil | ||
| 42 | return nil | ||
| 43 | } | ||
| 44 | 35 | ||
| 36 | // Get cache from redis. | ||
| 37 | func (rc *RedisCache) Get(key string) interface{} { | ||
| 38 | v, err := rc.do("HGET", rc.key, key) | ||
| 45 | if err != nil { | 39 | if err != nil { |
| 46 | return nil | 40 | return nil |
| 47 | } | 41 | } |
| ... | @@ -52,61 +46,19 @@ func (rc *RedisCache) Get(key string) interface{} { | ... | @@ -52,61 +46,19 @@ func (rc *RedisCache) Get(key string) interface{} { |
| 52 | // put cache to redis. | 46 | // put cache to redis. |
| 53 | // timeout is ignored. | 47 | // timeout is ignored. |
| 54 | func (rc *RedisCache) Put(key string, val interface{}, timeout int64) error { | 48 | func (rc *RedisCache) Put(key string, val interface{}, timeout int64) error { |
| 55 | if rc.c == nil { | 49 | _, err := rc.do("HSET", rc.key, key, val) |
| 56 | var err error | ||
| 57 | rc.c, err = rc.connectInit() | ||
| 58 | if err != nil { | ||
| 59 | return err | ||
| 60 | } | ||
| 61 | } | ||
| 62 | |||
| 63 | _, err := rc.c.Do("HSET", rc.key, key, val) | ||
| 64 | // write to closed socket, reset rc.c to nil | ||
| 65 | if err == io.EOF { | ||
| 66 | rc.c = nil | ||
| 67 | return err | ||
| 68 | } | ||
| 69 | |||
| 70 | return err | 50 | return err |
| 71 | } | 51 | } |
| 72 | 52 | ||
| 73 | // delete cache in redis. | 53 | // delete cache in redis. |
| 74 | func (rc *RedisCache) Delete(key string) error { | 54 | func (rc *RedisCache) Delete(key string) error { |
| 75 | if rc.c == nil { | 55 | _, err := rc.do("HDEL", rc.key, key) |
| 76 | var err error | ||
| 77 | rc.c, err = rc.connectInit() | ||
| 78 | if err != nil { | ||
| 79 | return err | ||
| 80 | } | ||
| 81 | } | ||
| 82 | |||
| 83 | _, err := rc.c.Do("HDEL", rc.key, key) | ||
| 84 | // write to closed socket, reset rc.c to nil | ||
| 85 | if err == io.EOF { | ||
| 86 | rc.c = nil | ||
| 87 | return err | ||
| 88 | } | ||
| 89 | |||
| 90 | return err | 56 | return err |
| 91 | } | 57 | } |
| 92 | 58 | ||
| 93 | // check cache exist in redis. | 59 | // check cache exist in redis. |
| 94 | func (rc *RedisCache) IsExist(key string) bool { | 60 | func (rc *RedisCache) IsExist(key string) bool { |
| 95 | if rc.c == nil { | 61 | v, err := redis.Bool(rc.do("HEXISTS", rc.key, key)) |
| 96 | var err error | ||
| 97 | rc.c, err = rc.connectInit() | ||
| 98 | if err != nil { | ||
| 99 | return false | ||
| 100 | } | ||
| 101 | } | ||
| 102 | |||
| 103 | v, err := redis.Bool(rc.c.Do("HEXISTS", rc.key, key)) | ||
| 104 | // write to closed socket, reset rc.c to nil | ||
| 105 | if err == io.EOF { | ||
| 106 | rc.c = nil | ||
| 107 | return false | ||
| 108 | } | ||
| 109 | |||
| 110 | if err != nil { | 62 | if err != nil { |
| 111 | return false | 63 | return false |
| 112 | } | 64 | } |
| ... | @@ -116,59 +68,19 @@ func (rc *RedisCache) IsExist(key string) bool { | ... | @@ -116,59 +68,19 @@ func (rc *RedisCache) IsExist(key string) bool { |
| 116 | 68 | ||
| 117 | // increase counter in redis. | 69 | // increase counter in redis. |
| 118 | func (rc *RedisCache) Incr(key string) error { | 70 | func (rc *RedisCache) Incr(key string) error { |
| 119 | if rc.c == nil { | 71 | _, err := redis.Bool(rc.do("HINCRBY", rc.key, key, 1)) |
| 120 | var err error | ||
| 121 | rc.c, err = rc.connectInit() | ||
| 122 | if err != nil { | ||
| 123 | return err | ||
| 124 | } | ||
| 125 | } | ||
| 126 | |||
| 127 | _, err := redis.Bool(rc.c.Do("HINCRBY", rc.key, key, 1)) | ||
| 128 | // write to closed socket | ||
| 129 | if err == io.EOF { | ||
| 130 | rc.c = nil | ||
| 131 | } | ||
| 132 | |||
| 133 | return err | 72 | return err |
| 134 | } | 73 | } |
| 135 | 74 | ||
| 136 | // decrease counter in redis. | 75 | // decrease counter in redis. |
| 137 | func (rc *RedisCache) Decr(key string) error { | 76 | func (rc *RedisCache) Decr(key string) error { |
| 138 | if rc.c == nil { | 77 | _, err := redis.Bool(rc.do("HINCRBY", rc.key, key, -1)) |
| 139 | var err error | ||
| 140 | rc.c, err = rc.connectInit() | ||
| 141 | if err != nil { | ||
| 142 | return err | ||
| 143 | } | ||
| 144 | } | ||
| 145 | |||
| 146 | _, err := redis.Bool(rc.c.Do("HINCRBY", rc.key, key, -1)) | ||
| 147 | |||
| 148 | // write to closed socket | ||
| 149 | if err == io.EOF { | ||
| 150 | rc.c = nil | ||
| 151 | } | ||
| 152 | |||
| 153 | return err | 78 | return err |
| 154 | } | 79 | } |
| 155 | 80 | ||
| 156 | // clean all cache in redis. delete this redis collection. | 81 | // clean all cache in redis. delete this redis collection. |
| 157 | func (rc *RedisCache) ClearAll() error { | 82 | func (rc *RedisCache) ClearAll() error { |
| 158 | if rc.c == nil { | 83 | _, err := rc.do("DEL", rc.key) |
| 159 | var err error | ||
| 160 | rc.c, err = rc.connectInit() | ||
| 161 | if err != nil { | ||
| 162 | return err | ||
| 163 | } | ||
| 164 | } | ||
| 165 | |||
| 166 | _, err := rc.c.Do("DEL", rc.key) | ||
| 167 | // write to closed socket | ||
| 168 | if err == io.EOF { | ||
| 169 | rc.c = nil | ||
| 170 | } | ||
| 171 | |||
| 172 | return err | 84 | return err |
| 173 | } | 85 | } |
| 174 | 86 | ||
| ... | @@ -179,32 +91,42 @@ func (rc *RedisCache) ClearAll() error { | ... | @@ -179,32 +91,42 @@ func (rc *RedisCache) ClearAll() error { |
| 179 | func (rc *RedisCache) StartAndGC(config string) error { | 91 | func (rc *RedisCache) StartAndGC(config string) error { |
| 180 | var cf map[string]string | 92 | var cf map[string]string |
| 181 | json.Unmarshal([]byte(config), &cf) | 93 | json.Unmarshal([]byte(config), &cf) |
| 94 | |||
| 182 | if _, ok := cf["key"]; !ok { | 95 | if _, ok := cf["key"]; !ok { |
| 183 | cf["key"] = DefaultKey | 96 | cf["key"] = DefaultKey |
| 184 | } | 97 | } |
| 98 | |||
| 185 | if _, ok := cf["conn"]; !ok { | 99 | if _, ok := cf["conn"]; !ok { |
| 186 | return errors.New("config has no conn key") | 100 | return errors.New("config has no conn key") |
| 187 | } | 101 | } |
| 102 | |||
| 188 | rc.key = cf["key"] | 103 | rc.key = cf["key"] |
| 189 | rc.conninfo = cf["conn"] | 104 | rc.conninfo = cf["conn"] |
| 190 | var err error | 105 | rc.connectInit() |
| 191 | rc.c, err = rc.connectInit() | 106 | |
| 192 | if err != nil { | 107 | c := rc.p.Get() |
| 108 | defer c.Close() | ||
| 109 | if err := c.Err(); err != nil { | ||
| 193 | return err | 110 | return err |
| 194 | } | 111 | } |
| 195 | if rc.c == nil { | 112 | |
| 196 | return errors.New("dial tcp conn error") | ||
| 197 | } | ||
| 198 | return nil | 113 | return nil |
| 199 | } | 114 | } |
| 200 | 115 | ||
| 201 | // connect to redis. | 116 | // connect to redis. |
| 202 | func (rc *RedisCache) connectInit() (redis.Conn, error) { | 117 | func (rc *RedisCache) connectInit() { |
| 118 | // initialize a new pool | ||
| 119 | rc.p = &redis.Pool{ | ||
| 120 | MaxIdle: 3, | ||
| 121 | IdleTimeout: 180 * time.Second, | ||
| 122 | Dial: func() (redis.Conn, error) { | ||
| 203 | c, err := redis.Dial("tcp", rc.conninfo) | 123 | c, err := redis.Dial("tcp", rc.conninfo) |
| 204 | if err != nil { | 124 | if err != nil { |
| 205 | return nil, err | 125 | return nil, err |
| 206 | } | 126 | } |
| 207 | return c, nil | 127 | return c, nil |
| 128 | }, | ||
| 129 | } | ||
| 208 | } | 130 | } |
| 209 | 131 | ||
| 210 | func init() { | 132 | func init() { | ... | ... |
-
Please register or sign in to post a comment