raftKit

Documentation

Overview

Package raftKit

Transport: raft集群内部节点之间的通信渠道，节点之间需要通过这个通道来进行日志同步、leader选举等.
(1) TCPTransport: 		基于tcp，可以跨机器跨网络通信.
(2) InmemTransport:		以内存为基础的传输层实现，它使用一个缓冲区来存存储请求和响应，并使用一个消费者通道来接收数据.
(3) NetworkTransport:	综合了TCPTransport和InmemTransport的优点的传输层实现，它既支持TCP协议提供的可靠性和高效性，又支持内存提供的简单性和灵活性.

Index

• Constants
• Variables
• func NewBoltStore(filePath string) (*raftboltdb.BoltStore, error)
• type RaftNode
  • func NewRaftNodeAndBootstrapCluster(addr string, nodeAddrs []string, dir string, fsm raft.FSM, logger hclog.Logger) (*RaftNode, error)
  • func (node *RaftNode) GetLeaderAddr() string
  • func (node *RaftNode) IsLeader() bool
  • func (node *RaftNode) IsLeader1() bool

Constants

const (
	// DefaultApplyTimeout raft.Apply() 默认的超时时间.
	DefaultApplyTimeout = time.Second * 3
)

Variables

var (
	// NewRaft 创建节点.
	/*
		@param conf 	*raft.Config		节点配置
		@param fsm 		raft.FSM			有限状态机（finite state machine）
		@param logs 	raft.LogStore		用来存储raft的日志
		@param stable 	raft.StableStore	稳定存储，用来存储raft集群的节点信息等
		@param snaps 	raft.SnapshotStore	快照存储，用来存储节点的快照信息
		@param trans 	raft.Transport		raft节点内部的通信通道
	*/
	NewRaft func(conf *raft.Config, fsm raft.FSM, logs raft.LogStore, stable raft.StableStore, snaps raft.SnapshotStore, trans raft.Transport) (*raft.Raft, error) = raft.NewRaft

	// RecoverCluster
	RecoverCluster func(conf *raft.Config, fsm raft.FSM, logs raft.LogStore, stable raft.StableStore, snaps raft.SnapshotStore, trans raft.Transport, configuration raft.Configuration) error = raft.RecoverCluster

	// BootstrapCluster
	BootstrapCluster func(conf *raft.Config, logs raft.LogStore, stable raft.StableStore, snaps raft.SnapshotStore, trans raft.Transport, configuration raft.Configuration) error = raft.BootstrapCluster

	// GetConfiguration
	GetConfiguration func(conf *raft.Config, fsm raft.FSM, logs raft.LogStore, stable raft.StableStore, snaps raft.SnapshotStore, trans raft.Transport) (raft.Configuration, error) = raft.GetConfiguration

	// HasExistingState
	HasExistingState func(logs raft.LogStore, stable raft.StableStore, snaps raft.SnapshotStore) (bool, error) = raft.HasExistingState
)

var (
	// NewDiscardSnapshotStore
	/*
		PS:
		(1) DiscardSnapshotStore: 不存储，忽略快照，相当于/dev/null（一般用于测试）
		(2) 实现了接口: raft.SnapshotStore

	*/
	NewDiscardSnapshotStore func() *raft.DiscardSnapshotStore = raft.NewDiscardSnapshotStore

	// NewFileSnapshotStore
	/*
		PS:
		(1) FileSnapshotStore: 文件持久化存储
		(2) 实现了接口: raft.SnapshotStore
	*/
	NewFileSnapshotStore           func(base string, retain int, logOutput io.Writer) (*raft.FileSnapshotStore, error) = raft.NewFileSnapshotStore
	NewFileSnapshotStoreWithLogger func(base string, retain int, logger hclog.Logger) (*raft.FileSnapshotStore, error) = raft.NewFileSnapshotStoreWithLogger

	// NewInmemSnapshotStore
	/*
		PS:
		(1) InmemSnapshotStore: 内存存储，不持久化，重启程序会丢失
		(2) 实现了接口: raft.SnapshotStore
	*/
	NewInmemSnapshotStore func() *raft.InmemSnapshotStore = raft.NewInmemSnapshotStore
)

var (
	NewTCPTransport func(bindAddr string, advertise net.Addr, maxPool int, timeout time.Duration, logOutput io.Writer) (*raft.NetworkTransport, error) = raft.NewTCPTransport

	NewTCPTransportWithConfig func(bindAddr string, advertise net.Addr, config *raft.NetworkTransportConfig) (*raft.NetworkTransport, error) = raft.NewTCPTransportWithConfig

	NewTCPTransportWithLogger func(bindAddr string, advertise net.Addr, maxPool int, timeout time.Duration, logger hclog.Logger) (*raft.NetworkTransport, error) = raft.NewTCPTransportWithLogger
)

var (
	NewInmemTransport            func(addr raft.ServerAddress) (raft.ServerAddress, *raft.InmemTransport)                        = raft.NewInmemTransport
	NewInmemTransportWithTimeout func(addr raft.ServerAddress, timeout time.Duration) (raft.ServerAddress, *raft.InmemTransport) = raft.NewInmemTransportWithTimeout
)

var (
	NewNetworkTransport           func(stream raft.StreamLayer, maxPool int, timeout time.Duration, logOutput io.Writer) *raft.NetworkTransport = raft.NewNetworkTransport
	NewNetworkTransportWithConfig func(config *raft.NetworkTransportConfig) *raft.NetworkTransport                                              = raft.NewNetworkTransportWithConfig
	NewNetworkTransportWithLogger func(stream raft.StreamLayer, maxPool int, timeout time.Duration, logger hclog.Logger) *raft.NetworkTransport = raft.NewNetworkTransportWithLogger
)

var (
	// DefaultConfig 默认的配置.
	DefaultConfig func() *raft.Config = raft.DefaultConfig
)

var (
	// NewInmemStore
	/*
		PS:
		(1) 实现了接口: raft.LogStore、raft.StableStore
	*/
	NewInmemStore func() *raft.InmemStore = raft.NewInmemStore
)

Functions

func NewBoltStore

func NewBoltStore(filePath string) (*raftboltdb.BoltStore, error)

NewBoltStore

PS:
(1) hashicorp提供了一个 raft-boltdb 来实现底层存储，它是一个嵌入式的数据库，能够 持久化 存储数据，我们直接用它来实现 LogStore 和 StableStore.
(2) 实现了接口: raft.LogStore、raft.StableStore

@param filePath string 存储路径（建议以".bolt"为文件后缀）

Types

type RaftNode

type RaftNode struct {
	*raft.Raft

	FSM raft.FSM
	// contains filtered or unexported fields
}

func NewRaftNodeAndBootstrapCluster

func NewRaftNodeAndBootstrapCluster(addr string, nodeAddrs []string, dir string, fsm raft.FSM, logger hclog.Logger) (*RaftNode, error)

NewRaftNodeAndBootstrapCluster

PS: 将 传参addr 作为id，所以传参中无id.

@param addr raft节点的地址（不能为""） @param nodeAddrs raft集群所有节点的地址（至少3个） @param dir raft节点的数据目录 @param fsm 不能为nil @param logger (1) raft节点的日志输出

(2) 可以为nil（将使用默认值: debug级别、输出到控制台）

func (*RaftNode) GetLeaderAddr

func (node *RaftNode) GetLeaderAddr() string

func (*RaftNode) IsLeader

func (node *RaftNode) IsLeader() bool

func (*RaftNode) IsLeader1

func (node *RaftNode) IsLeader1() bool

IsLeader1 当前Raft节点是否是 Leader ？

监听leader变化（使用此方法无法保证强一致性读，仅做leader变化过程观察）