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