re
/
redis
forked from mirror/redis
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Use PubSub.Channel to not burn CPU on errors

This commit is contained in:
Vladimir Mihailenco 2018-10-01 11:38:21 +03:00
parent 69445c6e87
commit 48e9afe2a7
3 changed files with 108 additions and 96 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
example_test.go
View File

@ -20,7 +20,6 @@ func init() {
PoolSize: 10,
PoolTimeout: 30 * time.Second,
})
// redisdb.FlushDB()
}
func ExampleNewClient() {

1
pubsub.go
View File

@ -51,7 +51,6 @@ func (c *PubSub) _conn(newChannels []string) (*pool.Conn, error) {
if c.closed {
return nil, pool.ErrClosed
}
if c.cn != nil {
return c.cn, nil
}

194
sentinel.go
View File

@ -119,7 +119,7 @@ func NewSentinelClient(opt *Options) *SentinelClient {
return c
}
func (c *SentinelClient) PubSub() *PubSub {
func (c *SentinelClient) pubSub() *PubSub {
pubsub := &PubSub{
opt: c.opt,
@ -132,14 +132,34 @@ func (c *SentinelClient) PubSub() *PubSub {
return pubsub
}
// Subscribe subscribes the client to the specified channels.
// Channels can be omitted to create empty subscription.
func (c *SentinelClient) Subscribe(channels ...string) *PubSub {
pubsub := c.pubSub()
if len(channels) > 0 {
_ = pubsub.Subscribe(channels...)
}
return pubsub
}
// PSubscribe subscribes the client to the given patterns.
// Patterns can be omitted to create empty subscription.
func (c *SentinelClient) PSubscribe(channels ...string) *PubSub {
pubsub := c.pubSub()
if len(channels) > 0 {
_ = pubsub.PSubscribe(channels...)
}
return pubsub
}
func (c *SentinelClient) GetMasterAddrByName(name string) *StringSliceCmd {
cmd := NewStringSliceCmd("SENTINEL", "get-master-addr-by-name", name)
cmd := NewStringSliceCmd("sentinel", "get-master-addr-by-name", name)
c.Process(cmd)
return cmd
}
func (c *SentinelClient) Sentinels(name string) *SliceCmd {
cmd := NewSliceCmd("SENTINEL", "sentinels", name)
cmd := NewSliceCmd("sentinel", "sentinels", name)
c.Process(cmd)
return cmd
}
@ -158,77 +178,70 @@ type sentinelFailover struct {
sentinel *SentinelClient
}
func (d *sentinelFailover) Close() error {
return d.resetSentinel()
func (c *sentinelFailover) Close() error {
c.mu.Lock()
defer c.mu.Unlock()
return c.closeSentinel()
}
func (d *sentinelFailover) Pool() *pool.ConnPool {
d.poolOnce.Do(func() {
d.opt.Dialer = d.dial
d.pool = newConnPool(d.opt)
func (c *sentinelFailover) Pool() *pool.ConnPool {
c.poolOnce.Do(func() {
c.opt.Dialer = c.dial
c.pool = newConnPool(c.opt)
})
return d.pool
return c.pool
}
func (d *sentinelFailover) dial() (net.Conn, error) {
addr, err := d.MasterAddr()
func (c *sentinelFailover) dial() (net.Conn, error) {
addr, err := c.MasterAddr()
if err != nil {
return nil, err
}
return net.DialTimeout("tcp", addr, d.opt.DialTimeout)
return net.DialTimeout("tcp", addr, c.opt.DialTimeout)
}
func (d *sentinelFailover) MasterAddr() (string, error) {
d.mu.Lock()
defer d.mu.Unlock()
addr, err := d.masterAddr()
func (c *sentinelFailover) MasterAddr() (string, error) {
addr, err := c.masterAddr()
if err != nil {
return "", err
}
d._switchMaster(addr)
c.switchMaster(addr)
return addr, nil
}
func (d *sentinelFailover) masterAddr() (string, error) {
// Try last working sentinel.
if d.sentinel != nil {
addr, err := d.sentinel.GetMasterAddrByName(d.masterName).Result()
if err == nil {
addr := net.JoinHostPort(addr[0], addr[1])
func (c *sentinelFailover) masterAddr() (string, error) {
addr := c.getMasterAddr()
if addr != "" {
return addr, nil
}
internal.Logf("sentinel: GetMasterAddrByName name=%q failed: %s",
d.masterName, err)
d._resetSentinel()
}
c.mu.Lock()
defer c.mu.Unlock()
for i, sentinelAddr := range d.sentinelAddrs {
for i, sentinelAddr := range c.sentinelAddrs {
sentinel := NewSentinelClient(&Options{
Addr: sentinelAddr,
DialTimeout: d.opt.DialTimeout,
ReadTimeout: d.opt.ReadTimeout,
WriteTimeout: d.opt.WriteTimeout,
DialTimeout: c.opt.DialTimeout,
ReadTimeout: c.opt.ReadTimeout,
WriteTimeout: c.opt.WriteTimeout,
PoolSize: d.opt.PoolSize,
PoolTimeout: d.opt.PoolTimeout,
IdleTimeout: d.opt.IdleTimeout,
PoolSize: c.opt.PoolSize,
PoolTimeout: c.opt.PoolTimeout,
IdleTimeout: c.opt.IdleTimeout,
})
masterAddr, err := sentinel.GetMasterAddrByName(d.masterName).Result()
masterAddr, err := sentinel.GetMasterAddrByName(c.masterName).Result()
if err != nil {
internal.Logf("sentinel: GetMasterAddrByName master=%q failed: %s",
d.masterName, err)
c.masterName, err)
sentinel.Close()
continue
}
// Push working sentinel to the top.
d.sentinelAddrs[0], d.sentinelAddrs[i] = d.sentinelAddrs[i], d.sentinelAddrs[0]
d.setSentinel(sentinel)
c.sentinelAddrs[0], c.sentinelAddrs[i] = c.sentinelAddrs[i], c.sentinelAddrs[0]
c.setSentinel(sentinel)
addr := net.JoinHostPort(masterAddr[0], masterAddr[1])
return addr, nil
@ -237,17 +250,41 @@ func (d *sentinelFailover) masterAddr() (string, error) {
return "", errors.New("redis: all sentinels are unreachable")
}
func (c *sentinelFailover) switchMaster(addr string) {
func (c *sentinelFailover) getMasterAddr() string {
c.mu.RLock()
sentinel := c.sentinel
c.mu.RUnlock()
if sentinel == nil {
return ""
}
addr, err := sentinel.GetMasterAddrByName(c.masterName).Result()
if err != nil {
internal.Logf("sentinel: GetMasterAddrByName name=%q failed: %s",
c.masterName, err)
c.mu.Lock()
c._switchMaster(addr)
if c.sentinel == sentinel {
c.closeSentinel()
}
c.mu.Unlock()
return ""
}
return net.JoinHostPort(addr[0], addr[1])
}
func (c *sentinelFailover) _switchMaster(addr string) {
if c._masterAddr == addr {
func (c *sentinelFailover) switchMaster(addr string) {
c.mu.RLock()
masterAddr := c._masterAddr
c.mu.RUnlock()
if masterAddr == addr {
return
}
c.mu.Lock()
defer c.mu.Unlock()
internal.Logf("sentinel: new master=%q addr=%q",
c.masterName, addr)
_ = c.Pool().Filter(func(cn *pool.Conn) bool {
@ -256,32 +293,22 @@ func (c *sentinelFailover) _switchMaster(addr string) {
c._masterAddr = addr
}
func (d *sentinelFailover) setSentinel(sentinel *SentinelClient) {
d.discoverSentinels(sentinel)
d.sentinel = sentinel
go d.listen(sentinel)
func (c *sentinelFailover) setSentinel(sentinel *SentinelClient) {
c.discoverSentinels(sentinel)
c.sentinel = sentinel
go c.listen(sentinel)
}
func (d *sentinelFailover) resetSentinel() error {
var err error
d.mu.Lock()
if d.sentinel != nil {
err = d._resetSentinel()
}
d.mu.Unlock()
func (c *sentinelFailover) closeSentinel() error {
err := c.sentinel.Close()
c.sentinel = nil
return err
}
func (d *sentinelFailover) _resetSentinel() error {
err := d.sentinel.Close()
d.sentinel = nil
return err
}
func (d *sentinelFailover) discoverSentinels(sentinel *SentinelClient) {
sentinels, err := sentinel.Sentinels(d.masterName).Result()
func (c *sentinelFailover) discoverSentinels(sentinel *SentinelClient) {
sentinels, err := sentinel.Sentinels(c.masterName).Result()
if err != nil {
internal.Logf("sentinel: Sentinels master=%q failed: %s", d.masterName, err)
internal.Logf("sentinel: Sentinels master=%q failed: %s", c.masterName, err)
return
}
for _, sentinel := range sentinels {
@ -290,49 +317,36 @@ func (d *sentinelFailover) discoverSentinels(sentinel *SentinelClient) {
key := vals[i].(string)
if key == "name" {
sentinelAddr := vals[i+1].(string)
if !contains(d.sentinelAddrs, sentinelAddr) {
internal.Logf(
"sentinel: discovered new sentinel=%q for master=%q",
sentinelAddr, d.masterName,
)
d.sentinelAddrs = append(d.sentinelAddrs, sentinelAddr)
if !contains(c.sentinelAddrs, sentinelAddr) {
internal.Logf("sentinel: discovered new sentinel=%q for master=%q",
sentinelAddr, c.masterName)
c.sentinelAddrs = append(c.sentinelAddrs, sentinelAddr)
}
}
}
}
}
func (d *sentinelFailover) listen(sentinel *SentinelClient) {
pubsub := sentinel.PubSub()
func (c *sentinelFailover) listen(sentinel *SentinelClient) {
pubsub := sentinel.Subscribe("+switch-master")
defer pubsub.Close()
err := pubsub.Subscribe("+switch-master")
if err != nil {
internal.Logf("sentinel: Subscribe failed: %s", err)
d.resetSentinel()
return
}
ch := pubsub.Channel()
for {
msg, err := pubsub.ReceiveMessage()
if err != nil {
if err == pool.ErrClosed {
d.resetSentinel()
return
}
internal.Logf("sentinel: ReceiveMessage failed: %s", err)
continue
msg, ok := <-ch
if !ok {
break
}
switch msg.Channel {
case "+switch-master":
parts := strings.Split(msg.Payload, " ")
if parts[0] != d.masterName {
if parts[0] != c.masterName {
internal.Logf("sentinel: ignore addr for master=%q", parts[0])
continue
}
addr := net.JoinHostPort(parts[3], parts[4])
d.switchMaster(addr)
c.switchMaster(addr)
}
}
}