基于hashicorp/raft的分布式一致性实战教学

type cacheManager struct {
        data map[string]string
        sync.RWMutex
}

// NewRaft is used to construct a new Raft node. It takes a configuration, as well
// as implementations of various interfaces that are required. If we have any
// old state, such as snapshots, logs, peers, etc, all those will be restored
// when creating the Raft node.
func NewRaft(conf *Config,
	fsm FSM,
	logs LogStore,
	stable StableStore,
	snaps SnapshotStore,
	trans Transport) (*Raft, error) {

    raftConfig := raft.DefaultConfig()
    raftConfig.LocalID = raft.ServerID(opts.raftTCPAddress)
    raftConfig.Logger = log.New(os.Stderr, "raft: ", log.Ldate|log.Ltime)

   logStore, err := raftboltdb.NewBoltStore(filepath.Join(opts.dataDir, 
                                            "raft-log.bolt"))                            
   stableStore, err := raftboltdb.NewBoltStore(filepath.Join(opts.dataDir, 
                                               "raft-stable.bolt"))                      

    snapshotStore, err := raft.NewFileSnapshotStore(opts.dataDir, 1, os.Stderr)

func newRaftTransport(opts *options) (*raft.NetworkTransport, error) {
    address, err := net.ResolveTCPAddr("tcp", opts.raftTCPAddress)
    if err != nil {
        return nil, err
    }
    transport, err := raft.NewTCPTransport(address.String(), address, 3, 10*time.Second, os.Stderr)
    if err != nil {
        return nil, err
    }
    return transport, nil
}

/*FSM provides an interface that can be implemented by
clients to make use of the replicated log.*/
type FSM interface {
	/* Apply log is invoked once a log entry is committed.
	It returns a value which will be made available in the
	ApplyFuture returned by Raft.Apply method if that
	method was called on the same Raft node as the FSM.*/
	Apply(*Log) interface{}
	// Snapshot is used to support log compaction. This call should
	// return an FSMSnapshot which can be used to save a point-in-time
	// snapshot of the FSM. Apply and Snapshot are not called in multiple
	// threads, but Apply will be called concurrently with Persist. This means
	// the FSM should be implemented in a fashion that allows for concurrent
	// updates while a snapshot is happening.
	Snapshot() (FSMSnapshot, error)
	// Restore is used to restore an FSM from a snapshot. It is not called
	// concurrently with any other command. The FSM must discard all previous
	// state.
	Restore(io.ReadCloser) error
}

// doSet saves data to cache, only raft master node provides this api
func (h *httpServer) doSet(w http.ResponseWriter, r *http.Request) {
    // ... get params from request url

    event := logEntryData{Key: key, Value: value}
    eventBytes, err := json.Marshal(event)
    if err != nil {
        h.log.Printf("json.Marshal failed, err:%v", err)
        fmt.Fprint(w, "internal error\n")
        return
    }

    applyFuture := h.ctx.st.raft.raft.Apply(eventBytes, 5*time.Second)
    if err := applyFuture.Error(); err != nil {
        h.log.Printf("raft.Apply failed:%v", err)
        fmt.Fprint(w, "internal error\n")
        return
    }

    fmt.Fprintf(w, "ok\n")
}

// Apply applies a Raft log entry to the key-value store.
func (f *FSM) Apply(logEntry *raft.Log) interface{} {
        e := logEntryData{}
        if err := json.Unmarshal(logEntry.Data, &e); err != nil {
                panic("Failed unmarshaling Raft log entry.")
        }
        ret := f.ctx.st.cm.Set(e.Key, e.Value)
        return ret
} 

// FSMSnapshot is returned by an FSM in response to a Snapshot
// It must be safe to invoke FSMSnapshot methods with concurrent
// calls to Apply.
type FSMSnapshot interface {
	// Persist should dump all necessary state to the WriteCloser 'sink',
	// and call sink.Close() when finished or call sink.Cancel() on error.
	Persist(sink SnapshotSink) error
	// Release is invoked when we are finished with the snapshot.
	Release()
}

type snapshot struct {
        cm *cacheManager
}
// Persist saves the FSM snapshot out to the given sink.
func (s *snapshot) Persist(sink raft.SnapshotSink) error {
        snapshotBytes, err := s.cm.Marshal()
        if err != nil {
                sink.Cancel()
                return err
        }
        if _, err := sink.Write(snapshotBytes); err != nil {
                sink.Cancel()
                return err
        }
        if err := sink.Close(); err != nil {
                sink.Cancel()
                return err
        }
        return nil
}
func (f *snapshot) Release() {}

    raftConfig := raft.DefaultConfig()
    raftConfig.LocalID = raft.ServerID(opts.raftTCPAddress)
    raftConfig.Logger = log.New(os.Stderr, "raft: ", log.Ldate|log.Ltime)
    raftConfig.SnapshotInterval = 20 * time.Second
    raftConfig.SnapshotThreshold = 2

// Restore stores the key-value store to a previous state.
func (f *FSM) Restore(serialized io.ReadCloser) error {
        return f.ctx.st.cm.UnMarshal(serialized)
}

// UnMarshal deserializes cache data
func (c *cacheManager) UnMarshal(serialized io.ReadCloser) error {
        var newData map[string]string
        if err := json.NewDecoder(serialized).Decode(&newData); err != nil {
                return err
        }
        c.Lock()
        defer c.Unlock()
        c.data = newData
        return nil
}

    if opts.bootstrap {
        configuration := raft.Configuration{
            Servers: []raft.Server{
                {
                    ID:      raftConfig.LocalID,
                    Address: transport.LocalAddr(),
                },
            },
        }
        raftNode.BootstrapCluster(configuration)
    }

// joinRaftCluster joins a node to raft cluster
func joinRaftCluster(opts *options) error {
    url := fmt.Sprintf("http://%s/join?peerAddress=%s", 
                       opts.joinAddress, 
                       opts.raftTCPAddress)
    resp, err := http.Get(url)
    if err != nil {
        return err
    }
    defer resp.Body.Close()
    body, err := ioutil.ReadAll(resp.Body)
    if err != nil {
        return err
    }
    if string(body) != "ok" {
        return errors.New(fmt.Sprintf("Error joining cluster: %s", body))
    }
    return nil
}

// doJoin handles joining cluster request
func (h *httpServer) doJoin(w http.ResponseWriter, r *http.Request) {
    vars := r.URL.Query()

    peerAddress := vars.Get("peerAddress")
    if peerAddress == "" {
        h.log.Println("invalid PeerAddress")
        fmt.Fprint(w, "invalid peerAddress\n")
        return
    }
    addPeerFuture := h.ctx.st.raft.raft.AddVoter(raft.ServerID(peerAddress), 
                                                 raft.ServerAddress(peerAddress), 
                                                 0, 0)
    if err := addPeerFuture.Error(); err != nil {
        h.log.Printf("Error joining peer to raft, peeraddress:%s, err:%v, code:%d", peerAddress, err, http.StatusInternalServerError)
        fmt.Fprint(w, "internal error\n")
        return
    }
    fmt.Fprint(w, "ok")
}

// doSet saves data to cache, only raft master node provides this api
func (h *httpServer) doSet(w http.ResponseWriter, r *http.Request) {
        if !h.checkWritePermission() {
                fmt.Fprint(w, "write method not allowed\n")
                return
        }
        // ... set data  
}

func newRaftNode(opts *options, ctx *stCachedContext) (*raftNodeInfo, error) {
    raftConfig := raft.DefaultConfig()
    raftConfig.LocalID = raft.ServerID(opts.raftTCPAddress)
    raftConfig.Logger = log.New(os.Stderr, "raft: ", log.Ldate|log.Ltime)
    raftConfig.SnapshotInterval = 20 * time.Second
    raftConfig.SnapshotThreshold = 2
    leaderNotifyCh := make(chan bool, 1)
    raftConfig.NotifyCh = leaderNotifyCh
    // ... 
}