1 Star 0 Fork 0

hlwqds/go-redis

加入 Gitee
与超过 1200万 开发者一起发现、参与优秀开源项目,私有仓库也完全免费 :)
免费加入
文件
克隆/下载
sentinel.go 7.11 KB
一键复制 编辑 原始数据 按行查看 历史
Vladimir Mihailenco 提交于 2016-03-12 12:41 . Optimize pool.Remove.
package redis
import (
"errors"
"fmt"
"net"
"strings"
"sync"
"time"
"gopkg.in/redis.v3/internal/pool"
)
//------------------------------------------------------------------------------
// FailoverOptions are used to configure a failover client and should
// be passed to NewFailoverClient.
type FailoverOptions struct {
// The master name.
MasterName string
// A seed list of host:port addresses of sentinel nodes.
SentinelAddrs []string
// Following options are copied from Options struct.
Password string
DB int64
DialTimeout time.Duration
ReadTimeout time.Duration
WriteTimeout time.Duration
PoolSize int
PoolTimeout time.Duration
IdleTimeout time.Duration
MaxRetries int
}
func (opt *FailoverOptions) options() *Options {
return &Options{
Addr: "FailoverClient",
DB: opt.DB,
Password: opt.Password,
DialTimeout: opt.DialTimeout,
ReadTimeout: opt.ReadTimeout,
WriteTimeout: opt.WriteTimeout,
PoolSize: opt.PoolSize,
PoolTimeout: opt.PoolTimeout,
IdleTimeout: opt.IdleTimeout,
MaxRetries: opt.MaxRetries,
}
}
// NewFailoverClient returns a Redis client that uses Redis Sentinel
// for automatic failover. It's safe for concurrent use by multiple
// goroutines.
func NewFailoverClient(failoverOpt *FailoverOptions) *Client {
opt := failoverOpt.options()
failover := &sentinelFailover{
masterName: failoverOpt.MasterName,
sentinelAddrs: failoverOpt.SentinelAddrs,
opt: opt,
}
base := baseClient{
opt: opt,
connPool: failover.Pool(),
onClose: func() error {
return failover.Close()
},
}
return &Client{
baseClient: base,
commandable: commandable{
process: base.process,
},
}
}
//------------------------------------------------------------------------------
type sentinelClient struct {
baseClient
commandable
}
func newSentinel(opt *Options) *sentinelClient {
base := baseClient{
opt: opt,
connPool: newConnPool(opt),
}
return &sentinelClient{
baseClient: base,
commandable: commandable{process: base.process},
}
}
func (c *sentinelClient) PubSub() *PubSub {
return &PubSub{
base: &baseClient{
opt: c.opt,
connPool: pool.NewStickyConnPool(c.connPool.(*pool.ConnPool), false),
},
}
}
func (c *sentinelClient) GetMasterAddrByName(name string) *StringSliceCmd {
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)
c.Process(cmd)
return cmd
}
type sentinelFailover struct {
masterName string
sentinelAddrs []string
opt *Options
pool *pool.ConnPool
poolOnce sync.Once
mu sync.RWMutex
sentinel *sentinelClient
}
func (d *sentinelFailover) Close() error {
return d.resetSentinel()
}
func (d *sentinelFailover) dial() (net.Conn, error) {
addr, err := d.MasterAddr()
if err != nil {
return nil, err
}
return net.DialTimeout("tcp", addr, d.opt.DialTimeout)
}
func (d *sentinelFailover) Pool() *pool.ConnPool {
d.poolOnce.Do(func() {
d.opt.Dialer = d.dial
d.pool = newConnPool(d.opt)
})
return d.pool
}
func (d *sentinelFailover) MasterAddr() (string, error) {
defer d.mu.Unlock()
d.mu.Lock()
// Try last working sentinel.
if d.sentinel != nil {
addr, err := d.sentinel.GetMasterAddrByName(d.masterName).Result()
if err != nil {
Logger.Printf("sentinel: GetMasterAddrByName %q failed: %s", d.masterName, err)
d._resetSentinel()
} else {
addr := net.JoinHostPort(addr[0], addr[1])
Logger.Printf("sentinel: %q addr is %s", d.masterName, addr)
return addr, nil
}
}
for i, sentinelAddr := range d.sentinelAddrs {
sentinel := newSentinel(&Options{
Addr: sentinelAddr,
DialTimeout: d.opt.DialTimeout,
ReadTimeout: d.opt.ReadTimeout,
WriteTimeout: d.opt.WriteTimeout,
PoolSize: d.opt.PoolSize,
PoolTimeout: d.opt.PoolTimeout,
IdleTimeout: d.opt.IdleTimeout,
})
masterAddr, err := sentinel.GetMasterAddrByName(d.masterName).Result()
if err != nil {
Logger.Printf("sentinel: GetMasterAddrByName %q failed: %s", d.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)
addr := net.JoinHostPort(masterAddr[0], masterAddr[1])
Logger.Printf("sentinel: %q addr is %s", d.masterName, addr)
return addr, nil
}
return "", errors.New("redis: all sentinels are unreachable")
}
func (d *sentinelFailover) setSentinel(sentinel *sentinelClient) {
d.discoverSentinels(sentinel)
d.sentinel = sentinel
go d.listen()
}
func (d *sentinelFailover) resetSentinel() error {
d.mu.Lock()
err := d._resetSentinel()
d.mu.Unlock()
return err
}
func (d *sentinelFailover) _resetSentinel() error {
var err error
if d.sentinel != nil {
err = d.sentinel.Close()
d.sentinel = nil
}
return err
}
func (d *sentinelFailover) discoverSentinels(sentinel *sentinelClient) {
sentinels, err := sentinel.Sentinels(d.masterName).Result()
if err != nil {
Logger.Printf("sentinel: Sentinels %q failed: %s", d.masterName, err)
return
}
for _, sentinel := range sentinels {
vals := sentinel.([]interface{})
for i := 0; i < len(vals); i += 2 {
key := vals[i].(string)
if key == "name" {
sentinelAddr := vals[i+1].(string)
if !contains(d.sentinelAddrs, sentinelAddr) {
Logger.Printf(
"sentinel: discovered new %q sentinel: %s",
d.masterName, sentinelAddr,
)
d.sentinelAddrs = append(d.sentinelAddrs, sentinelAddr)
}
}
}
}
}
// closeOldConns closes connections to the old master after failover switch.
func (d *sentinelFailover) closeOldConns(newMaster string) {
// Good connections that should be put back to the pool. They
// can't be put immediately, because pool.First will return them
// again on next iteration.
cnsToPut := make([]*pool.Conn, 0)
for {
cn := d.pool.First()
if cn == nil {
break
}
if cn.RemoteAddr().String() != newMaster {
err := fmt.Errorf(
"sentinel: closing connection to the old master %s",
cn.RemoteAddr(),
)
Logger.Print(err)
d.pool.Replace(cn, err)
} else {
cnsToPut = append(cnsToPut, cn)
}
}
for _, cn := range cnsToPut {
d.pool.Put(cn)
}
}
func (d *sentinelFailover) listen() {
var pubsub *PubSub
for {
if pubsub == nil {
pubsub = d.sentinel.PubSub()
if err := pubsub.Subscribe("+switch-master"); err != nil {
Logger.Printf("sentinel: Subscribe failed: %s", err)
d.resetSentinel()
return
}
}
msg, err := pubsub.ReceiveMessage()
if err != nil {
Logger.Printf("sentinel: ReceiveMessage failed: %s", err)
pubsub.Close()
d.resetSentinel()
return
}
switch msg.Channel {
case "+switch-master":
parts := strings.Split(msg.Payload, " ")
if parts[0] != d.masterName {
Logger.Printf("sentinel: ignore new %s addr", parts[0])
continue
}
addr := net.JoinHostPort(parts[3], parts[4])
Logger.Printf(
"sentinel: new %q addr is %s",
d.masterName, addr,
)
d.closeOldConns(addr)
}
}
}
func contains(slice []string, str string) bool {
for _, s := range slice {
if s == str {
return true
}
}
return false
}
Loading...
马建仓 AI 助手
尝试更多
代码解读
代码找茬
代码优化
1
https://gitee.com/hlwqds/go-redis.git
git@gitee.com:hlwqds/go-redis.git
hlwqds
go-redis
go-redis
v3.5.3

搜索帮助

0d507c66 1850385 C8b1a773 1850385