-
Notifications
You must be signed in to change notification settings - Fork 0
/
pool.go
87 lines (71 loc) · 1.6 KB
/
pool.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
package asredis
import (
"sync"
// "time"
//"fmt"
)
const (
defaultPoolSize int32 = 5
defaultQueueSize int32 = 1024 * 8
)
// 用来保存连接至单个redis进程的多个连接
type Pool struct {
Spec
queueChan chan *RequestsPkg
//连接个数
PoolSize int32
//发送请求到连接的队列长度
QueueSize int32
stopChan chan struct{}
stopWg sync.WaitGroup
}
func (p *Pool) Call(cmd string, args ...interface{}) (interface{}, error) {
reqPkg := NewRequstPkg()
reqPkg.Add(cmd, args...)
p.Pipelining(reqPkg)
req := reqPkg.requests[0]
return req.Reply, req.Err
}
func (p *Pool) Pipelining(reqPkg *RequestsPkg) {
p.queueChan <- reqPkg
reqPkg.wait()
}
func (p *Pool) Start() {
if p.stopChan != nil {
panic("asredis: the given client is already started. Call Stop() before calling Start()")
}
p.queueChan = make(chan *RequestsPkg, p.QueueSize)
p.stopChan = make(chan struct{})
for i := 0; i < p.PoolSize; i++ {
c := new(Connection)
go c.process(p)
}
}
func (p *Pool) Stop() {
if p.stopChan == nil {
panic("asredis: the client must be started before stopping it")
}
close(p.stopChan)
p.stopWg.Wait()
p.stopChan = nil
}
func (p *Pool) Eval(l *LuaEval, args ...interface{}) (reply interface{}, err error) {
reply, err = p.Call("EVALSHA", JoinArgs(l.hash, args)...)
if err != nil {
var content string
content, err = l.readFile()
if err != nil {
return
}
reply, err = p.Call("EVAL", JoinArgs(content, args)...)
}
return
}
func NewPool() *Pool {
pool := &Pool{
Spec: *DefaultSpec(),
PoolSize: defaultPoolSize,
QueueSize: defaultQueueSize,
}
return pool
}