代码拉取完成,页面将自动刷新
package redis
import (
"math/rand"
"sync"
"sync/atomic"
"time"
"gopkg.in/redis.v4/internal"
"gopkg.in/redis.v4/internal/errors"
"gopkg.in/redis.v4/internal/hashtag"
"gopkg.in/redis.v4/internal/pool"
)
type clusterNode struct {
Client *Client
Latency time.Duration
}
// ClusterClient is a Redis Cluster client representing a pool of zero
// or more underlying connections. It's safe for concurrent use by
// multiple goroutines.
type ClusterClient struct {
cmdable
opt *ClusterOptions
mu sync.RWMutex
addrs []string
nodes map[string]*clusterNode
slots [][]*clusterNode
closed bool
cmdsInfo map[string]*CommandInfo
cmdsInfoOnce *sync.Once
// Reports where slots reloading is in progress.
reloading uint32
}
var _ Cmdable = (*ClusterClient)(nil)
// NewClusterClient returns a Redis Cluster client as described in
// http://redis.io/topics/cluster-spec.
func NewClusterClient(opt *ClusterOptions) *ClusterClient {
opt.init()
c := &ClusterClient{
opt: opt,
nodes: make(map[string]*clusterNode),
cmdsInfoOnce: new(sync.Once),
}
c.cmdable.process = c.Process
for _, addr := range opt.Addrs {
_, _ = c.nodeByAddr(addr)
}
c.reloadSlots()
if opt.IdleCheckFrequency > 0 {
go c.reaper(opt.IdleCheckFrequency)
}
return c
}
func (c *ClusterClient) cmdInfo(name string) *CommandInfo {
c.cmdsInfoOnce.Do(func() {
for _, node := range c.nodes {
cmdsInfo, err := node.Client.Command().Result()
if err == nil {
c.cmdsInfo = cmdsInfo
return
}
}
c.cmdsInfoOnce = &sync.Once{}
})
return c.cmdsInfo[name]
}
func (c *ClusterClient) getNodes() map[string]*clusterNode {
var nodes map[string]*clusterNode
c.mu.RLock()
if !c.closed {
nodes = make(map[string]*clusterNode, len(c.nodes))
for addr, node := range c.nodes {
nodes[addr] = node
}
}
c.mu.RUnlock()
return nodes
}
func (c *ClusterClient) Watch(fn func(*Tx) error, keys ...string) error {
node, err := c.slotMasterNode(hashtag.Slot(keys[0]))
if err != nil {
return err
}
return node.Client.Watch(fn, keys...)
}
// PoolStats returns accumulated connection pool stats.
func (c *ClusterClient) PoolStats() *PoolStats {
var acc PoolStats
for _, node := range c.getNodes() {
s := node.Client.connPool.Stats()
acc.Requests += s.Requests
acc.Hits += s.Hits
acc.Timeouts += s.Timeouts
acc.TotalConns += s.TotalConns
acc.FreeConns += s.FreeConns
}
return &acc
}
// Close closes the cluster client, releasing any open resources.
//
// It is rare to Close a ClusterClient, as the ClusterClient is meant
// to be long-lived and shared between many goroutines.
func (c *ClusterClient) Close() error {
c.mu.Lock()
if !c.closed {
c.closeClients()
c.addrs = nil
c.nodes = nil
c.slots = nil
c.cmdsInfo = nil
}
c.closed = true
c.mu.Unlock()
return nil
}
func (c *ClusterClient) nodeByAddr(addr string) (*clusterNode, error) {
c.mu.RLock()
node, ok := c.nodes[addr]
c.mu.RUnlock()
if ok {
return node, nil
}
defer c.mu.Unlock()
c.mu.Lock()
if c.closed {
return nil, pool.ErrClosed
}
node, ok = c.nodes[addr]
if !ok {
node = c.newNode(addr)
c.nodes[addr] = node
c.addrs = append(c.addrs, addr)
}
return node, nil
}
func (c *ClusterClient) newNode(addr string) *clusterNode {
opt := c.opt.clientOptions()
opt.Addr = addr
return &clusterNode{
Client: NewClient(opt),
}
}
func (c *ClusterClient) slotNodes(slot int) (nodes []*clusterNode) {
c.mu.RLock()
if slot < len(c.slots) {
nodes = c.slots[slot]
}
c.mu.RUnlock()
return nodes
}
// randomNode returns random live node.
func (c *ClusterClient) randomNode() (*clusterNode, error) {
var nodeErr error
for i := 0; i < 10; i++ {
c.mu.RLock()
closed := c.closed
addrs := c.addrs
c.mu.RUnlock()
if closed {
return nil, pool.ErrClosed
}
n := rand.Intn(len(addrs))
node, err := c.nodeByAddr(addrs[n])
if err != nil {
return nil, err
}
nodeErr = node.Client.ClusterInfo().Err()
if nodeErr == nil {
return node, nil
}
}
return nil, nodeErr
}
func (c *ClusterClient) slotMasterNode(slot int) (*clusterNode, error) {
nodes := c.slotNodes(slot)
if len(nodes) == 0 {
return c.randomNode()
}
return nodes[0], nil
}
func (c *ClusterClient) slotSlaveNode(slot int) (*clusterNode, error) {
nodes := c.slotNodes(slot)
switch len(nodes) {
case 0:
return c.randomNode()
case 1:
return nodes[0], nil
case 2:
return nodes[1], nil
default:
n := rand.Intn(len(nodes)-1) + 1
return nodes[n], nil
}
}
func (c *ClusterClient) slotClosestNode(slot int) (*clusterNode, error) {
nodes := c.slotNodes(slot)
if len(nodes) == 0 {
return c.randomNode()
}
var node *clusterNode
for _, n := range nodes {
if node == nil || n.Latency < node.Latency {
node = n
}
}
return node, nil
}
func (c *ClusterClient) cmdSlotAndNode(cmd Cmder) (int, *clusterNode, error) {
cmdInfo := c.cmdInfo(cmd.arg(0))
if cmdInfo == nil {
internal.Logf("info for cmd=%s not found", cmd.arg(0))
node, err := c.randomNode()
return 0, node, err
}
if cmdInfo.FirstKeyPos == -1 {
node, err := c.randomNode()
return 0, node, err
}
firstKey := cmd.arg(int(cmdInfo.FirstKeyPos))
slot := hashtag.Slot(firstKey)
if cmdInfo.ReadOnly && c.opt.ReadOnly {
if c.opt.RouteByLatency {
node, err := c.slotClosestNode(slot)
return slot, node, err
}
node, err := c.slotSlaveNode(slot)
return slot, node, err
}
node, err := c.slotMasterNode(slot)
return slot, node, err
}
func (c *ClusterClient) Process(cmd Cmder) error {
slot, node, err := c.cmdSlotAndNode(cmd)
if err != nil {
cmd.setErr(err)
return err
}
var ask bool
for attempt := 0; attempt <= c.opt.MaxRedirects; attempt++ {
if attempt > 0 {
cmd.reset()
}
if ask {
pipe := node.Client.Pipeline()
pipe.Process(NewCmd("ASKING"))
pipe.Process(cmd)
_, _ = pipe.Exec()
pipe.Close()
ask = false
} else {
node.Client.Process(cmd)
}
// If there is no (real) error, we are done!
err := cmd.Err()
if err == nil {
return nil
}
// On network errors try random node.
if errors.IsRetryable(err) {
node, err = c.randomNode()
continue
}
var moved bool
var addr string
moved, ask, addr = errors.IsMoved(err)
if moved || ask {
master, _ := c.slotMasterNode(slot)
if moved && (master == nil || master.Client.getAddr() != addr) {
c.lazyReloadSlots()
}
node, err = c.nodeByAddr(addr)
if err != nil {
cmd.setErr(err)
return err
}
continue
}
break
}
return cmd.Err()
}
// ForEachMaster concurrently calls the fn on each master node in the cluster.
// It returns the first error if any.
func (c *ClusterClient) ForEachMaster(fn func(client *Client) error) error {
c.mu.RLock()
slots := c.slots
c.mu.RUnlock()
var wg sync.WaitGroup
visited := make(map[*clusterNode]struct{})
errCh := make(chan error, 1)
for _, nodes := range slots {
if len(nodes) == 0 {
continue
}
master := nodes[0]
if _, ok := visited[master]; ok {
continue
}
visited[master] = struct{}{}
wg.Add(1)
go func(node *clusterNode) {
defer wg.Done()
err := fn(node.Client)
if err != nil {
select {
case errCh <- err:
default:
}
}
}(master)
}
wg.Wait()
select {
case err := <-errCh:
return err
default:
return nil
}
}
// closeClients closes all clients and returns the first error if there are any.
func (c *ClusterClient) closeClients() error {
var retErr error
for _, node := range c.nodes {
if err := node.Client.Close(); err != nil && retErr == nil {
retErr = err
}
}
return retErr
}
func (c *ClusterClient) setSlots(cs []ClusterSlot) {
slots := make([][]*clusterNode, hashtag.SlotNumber)
for _, s := range cs {
var nodes []*clusterNode
for _, n := range s.Nodes {
node, err := c.nodeByAddr(n.Addr)
if err == nil {
nodes = append(nodes, node)
}
}
for i := s.Start; i <= s.End; i++ {
slots[i] = nodes
}
}
c.mu.Lock()
if !c.closed {
c.slots = slots
}
c.mu.Unlock()
}
func (c *ClusterClient) lazyReloadSlots() {
if !atomic.CompareAndSwapUint32(&c.reloading, 0, 1) {
return
}
go c.reloadSlots()
}
func (c *ClusterClient) reloadSlots() {
defer atomic.StoreUint32(&c.reloading, 0)
node, err := c.randomNode()
if err != nil {
return
}
slots, err := node.Client.ClusterSlots().Result()
if err != nil {
internal.Logf("ClusterSlots on addr=%q failed: %s", node.Client.getAddr(), err)
return
}
c.setSlots(slots)
if c.opt.RouteByLatency {
c.setNodesLatency()
}
}
func (c *ClusterClient) setNodesLatency() {
const n = 10
for _, node := range c.getNodes() {
var latency time.Duration
for i := 0; i < n; i++ {
t1 := time.Now()
node.Client.Ping()
latency += time.Since(t1)
}
node.Latency = latency / n
}
}
// reaper closes idle connections to the cluster.
func (c *ClusterClient) reaper(idleCheckFrequency time.Duration) {
ticker := time.NewTicker(idleCheckFrequency)
defer ticker.Stop()
for _ = range ticker.C {
nodes := c.getNodes()
if nodes == nil {
break
}
var n int
for _, node := range nodes {
nn, err := node.Client.connPool.(*pool.ConnPool).ReapStaleConns()
if err != nil {
internal.Logf("ReapStaleConns failed: %s", err)
} else {
n += nn
}
}
s := c.PoolStats()
internal.Logf(
"reaper: removed %d stale conns (TotalConns=%d FreeConns=%d Requests=%d Hits=%d Timeouts=%d)",
n, s.TotalConns, s.FreeConns, s.Requests, s.Hits, s.Timeouts,
)
}
}
func (c *ClusterClient) Pipeline() *Pipeline {
pipe := Pipeline{
exec: c.pipelineExec,
}
pipe.cmdable.process = pipe.Process
pipe.statefulCmdable.process = pipe.Process
return &pipe
}
func (c *ClusterClient) Pipelined(fn func(*Pipeline) error) ([]Cmder, error) {
return c.Pipeline().pipelined(fn)
}
func (c *ClusterClient) pipelineExec(cmds []Cmder) error {
var retErr error
setRetErr := func(err error) {
if retErr == nil {
retErr = err
}
}
cmdsMap := make(map[*clusterNode][]Cmder)
for _, cmd := range cmds {
_, node, err := c.cmdSlotAndNode(cmd)
if err != nil {
cmd.setErr(err)
setRetErr(err)
continue
}
cmdsMap[node] = append(cmdsMap[node], cmd)
}
for attempt := 0; attempt <= c.opt.MaxRedirects; attempt++ {
failedCmds := make(map[*clusterNode][]Cmder)
for node, cmds := range cmdsMap {
if node == nil {
var err error
node, err = c.randomNode()
if err != nil {
setCmdsErr(cmds, err)
setRetErr(err)
continue
}
}
cn, _, err := node.Client.conn()
if err != nil {
setCmdsErr(cmds, err)
setRetErr(err)
continue
}
failedCmds, err = c.execClusterCmds(cn, cmds, failedCmds)
if err != nil {
setRetErr(err)
}
node.Client.putConn(cn, err, false)
}
cmdsMap = failedCmds
}
return retErr
}
func (c *ClusterClient) execClusterCmds(
cn *pool.Conn, cmds []Cmder, failedCmds map[*clusterNode][]Cmder,
) (map[*clusterNode][]Cmder, error) {
if err := writeCmd(cn, cmds...); err != nil {
setCmdsErr(cmds, err)
return failedCmds, err
}
var retErr error
setRetErr := func(err error) {
if retErr == nil {
retErr = err
}
}
for i, cmd := range cmds {
err := cmd.readReply(cn)
if err == nil {
continue
}
if errors.IsNetwork(err) {
cmd.reset()
failedCmds[nil] = append(failedCmds[nil], cmds[i:]...)
break
} else if moved, ask, addr := errors.IsMoved(err); moved {
c.lazyReloadSlots()
cmd.reset()
node, err := c.nodeByAddr(addr)
if err != nil {
setRetErr(err)
continue
}
failedCmds[node] = append(failedCmds[node], cmd)
} else if ask {
cmd.reset()
node, err := c.nodeByAddr(addr)
if err != nil {
setRetErr(err)
continue
}
failedCmds[node] = append(failedCmds[node], NewCmd("ASKING"), cmd)
} else {
setRetErr(err)
}
}
return failedCmds, retErr
}
//------------------------------------------------------------------------------
// ClusterOptions are used to configure a cluster client and should be
// passed to NewClusterClient.
type ClusterOptions struct {
// A seed list of host:port addresses of cluster nodes.
Addrs []string
// The maximum number of retries before giving up. Command is retried
// on network errors and MOVED/ASK redirects.
// Default is 16.
MaxRedirects int
// Enables read queries for a connection to a Redis Cluster slave node.
ReadOnly bool
// Enables routing read-only queries to the closest master or slave node.
RouteByLatency bool
// Following options are copied from Options struct.
Password string
DialTimeout time.Duration
ReadTimeout time.Duration
WriteTimeout time.Duration
// PoolSize applies per cluster node and not for the whole cluster.
PoolSize int
PoolTimeout time.Duration
IdleTimeout time.Duration
IdleCheckFrequency time.Duration
}
func (opt *ClusterOptions) init() {
if opt.MaxRedirects == -1 {
opt.MaxRedirects = 0
} else if opt.MaxRedirects == 0 {
opt.MaxRedirects = 16
}
if opt.RouteByLatency {
opt.ReadOnly = true
}
}
func (opt *ClusterOptions) clientOptions() *Options {
const disableIdleCheck = -1
return &Options{
Password: opt.Password,
ReadOnly: opt.ReadOnly,
DialTimeout: opt.DialTimeout,
ReadTimeout: opt.ReadTimeout,
WriteTimeout: opt.WriteTimeout,
PoolSize: opt.PoolSize,
PoolTimeout: opt.PoolTimeout,
IdleTimeout: opt.IdleTimeout,
// IdleCheckFrequency is not copied to disable reaper
IdleCheckFrequency: disableIdleCheck,
}
}
此处可能存在不合适展示的内容,页面不予展示。您可通过相关编辑功能自查并修改。
如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容,可点击提交进行申诉,我们将尽快为您处理。