func (s *KingbusServer) Stop() {
	if s.started.Load() == false {
		return
	}
	s.adminSvr.Stop()
	s.prometheusSvr.Stop()
	s.StopServer(config.SyncerServerType)
	s.StopServer(config.BinlogServerType)
	s.raftNode.Stop()
	close(s.stopping)
	s.started.Store(false)
	return
}
Stop kingbus server
package server
import (
	"context"
	"fmt"
	"net/url"
	"time"
	"errors"
	"encoding/json"
	"sync"
	"strings"
	"github.com/coreos/etcd/etcdserver/stats"
	"github.com/coreos/etcd/pkg/idutil"
	"github.com/coreos/etcd/pkg/transport"
	"github.com/coreos/etcd/pkg/types"
	"github.com/coreos/etcd/pkg/wait"
	etcdraft "github.com/coreos/etcd/raft"
	"github.com/coreos/etcd/raft/raftpb"
	"github.com/coreos/etcd/rafthttp"
	"github.com/flike/kingbus/api"
	"github.com/flike/kingbus/config"
	"github.com/flike/kingbus/log"
	"github.com/flike/kingbus/mysql"
	"github.com/flike/kingbus/raft"
	"github.com/flike/kingbus/raft/membership"
	"github.com/flike/kingbus/storage"
	"github.com/flike/kingbus/utils"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/rcrowley/go-metrics"
	"go.uber.org/atomic"
)
const (
	maxInflightMsgs = 512
	HealthInterval = 5 * time.Second
	DialTimeout = 5 * time.Second
	ProposeMaxRetryCount = 1000
)
type KingbusServer struct {
	Cfg            *config.KingbusServerConfig
	adminSvr       *api.AdminServer
	syncer         *Syncer
	master         *BinlogServer
	binlogProgress *BinlogProgress
	prometheusSvr  *PrometheusServer
	id       types.ID
	raftNode *raft.Node
	cluster  *membership.RaftCluster
	store    storage.Storage
	lead            *atomic.Uint64
	leadElectedTime time.Time
	Mu     sync.RWMutex
	errorc chan error
	wait     wait.Wait
	reqIDGen *idutil.Generator
	applyBroadcast *utils.Broadcast
	stats  *stats.ServerStats
	lstats *stats.LeaderStats
	stopping chan struct{}
	started  *atomic.Bool
}
func (s *KingbusServer) starRaft(cfg config.RaftNodeConfig) error {
	var (
		etcdRaftNode etcdraft.Node
		id           types.ID
		cl           *membership.RaftCluster
		remotes      []*membership.Member
		appliedIndex uint64
	)
	prt, err := rafthttp.NewRoundTripper(transport.TLSInfo{}, DialTimeout)
	if err != nil {
		return err
	}
	store, err := storage.NewDiskStorage(cfg.DataDir, cfg.ReserveDataSize)
	if err != nil {
		log.Log.Fatalf("NewKingbusServer:NewDiskStorage error,err:%s,dir:%s", err.Error(), cfg.DataDir)
	}
	defer func() {
		if err != nil {
			store.Close()
		}
	}()
	logExist := utils.ExistLog(cfg.DataDir)
	switch {
	case !logExist && !cfg.NewCluster:
		if err = cfg.VerifyJoinExisting(); err != nil {
			return err
		}
		cl, err = membership.NewClusterFromURLsMap(cfg.InitialPeerURLsMap)
		if err != nil {
			return err
		}
		remotePeerURLs := membership.GetRemotePeerURLs(cl, cfg.Name)
		existingCluster, gerr := membership.GetClusterFromRemotePeers(remotePeerURLs, prt)
		if gerr != nil {
			return fmt.Errorf("cannot fetch cluster info from peer urls: %v", gerr)
		}
		if err = membership.ValidateClusterAndAssignIDs(cl, existingCluster); err != nil {
			return fmt.Errorf("error validating peerURLs %s: %v", existingCluster, err)
		}
		remotes = existingCluster.Members()
		cl.SetID(existingCluster.GetID())
		cl.SetStore(store)
		id, etcdRaftNode = startEtcdRaftNode(cfg, store, cl, nil)
	case !logExist && cfg.NewCluster:
		if err = cfg.VerifyBootstrap(); err != nil {
			return err
		}
		cl, err = membership.NewClusterFromURLsMap(cfg.InitialPeerURLsMap)
		if err != nil {
			return err
		}
		m := cl.MemberByName(cfg.Name)
		if membership.IsMemberBootstrapped(cl, cfg.Name, prt, DialTimeout) {
			return fmt.Errorf("member %s has already been bootstrapped", m.ID)
		}
		cl.SetStore(store)
		id, etcdRaftNode = startEtcdRaftNode(cfg, store, cl, cl.MemberIDs())
	case logExist:
		if err = utils.IsDirWriteable(cfg.DataDir); err != nil {
			return fmt.Errorf("cannot write to member directory: %v", err)
		}
		appliedIndex = raft.MustGetAppliedIndex(store)
		cfg.AppliedIndex = appliedIndex
		id, etcdRaftNode, cl = restartEtcdNode(cfg, store)
		cl.SetStore(store)
	default:
		return fmt.Errorf("unsupported bootstrap config")
	}
	s.raftNode = raft.NewNode(
		raft.NodeConfig{
			IsIDRemoved: func(id uint64) bool { return cl.IsIDRemoved(types.ID(id)) },
			Node:        etcdRaftNode,
			Heartbeat:   cfg.HeartbeatMs,
			Storage:     store,
		},
	)
	s.raftNode.SetAppliedIndex(appliedIndex)
	s.id = id
	s.wait = wait.New()
	s.reqIDGen = idutil.NewGenerator(uint16(id), time.Now())
	s.stopping = make(chan struct{})
	s.errorc = make(chan error)
	s.applyBroadcast = utils.NewBroadcast()
	s.stats = stats.NewServerStats(cfg.Name, id.String())
	s.lstats = stats.NewLeaderStats(id.String())
	s.store = store
	tr := &rafthttp.Transport{
		TLSInfo:     transport.TLSInfo{},
		DialTimeout: DialTimeout,
		ID:          id,
		URLs:        cfg.PeerURLs,
		ClusterID:   cl.GetID(),
		Raft:        s,
		ServerStats: s.stats,
		LeaderStats: s.lstats,
		ErrorC:      s.errorc,
	}
	if err = tr.Start(); err != nil {
		return err
	}
	for _, m := range remotes {
		if m.ID != id {
			tr.AddRemote(m.ID, m.PeerURLs)
		}
	}
	for _, m := range cl.Members() {
		if m.ID != id {
			tr.AddPeer(m.ID, m.PeerURLs)
		}
	}
	s.raftNode.Transport = tr
	s.cluster = cl
	return nil
}
func (s *KingbusServer) startRaftPeer(peerURLs types.URLs) error {
	err := s.raftNode.NewPeerListener(peerURLs)
	if err != nil {
		return err
	}
	s.raftNode.SetPeerHandler()
	s.raftNode.PeerHandlerServe()
	log.Log.Infof("startRaftPeer success")
	return nil
}
func (s *KingbusServer) startAdminServer(urls types.URLs) error {
	if len(urls) != 1 {
		return ErrArgs
	}
	addr := urls[0].Host
	s.adminSvr = api.NewAdminServer(addr, s, s.cluster)
	return nil
}
func (s *KingbusServer) startPrometheus(addr string) error {
	if len(addr) == 0 {
		log.Log.Errorf("metrics addr is nil")
		return ErrArgs
	}
	s.prometheusSvr = NewPrometheusServer(addr, metrics.DefaultRegistry, prometheus.DefaultRegisterer, time.Second)
	return nil
}
func NewKingbusServer(cfg *config.KingbusServerConfig) (*KingbusServer, error) {
	s := new(KingbusServer)
	s.Cfg = cfg
	s.lead = atomic.NewUint64(0)
	err := s.starRaft(s.Cfg.RaftNodeCfg)
	if err != nil {
		log.Log.Errorf("NewKingbusServer:startRaft error,err:%s", err)
		return nil, err
	}
	err = s.startRaftPeer(s.Cfg.RaftNodeCfg.PeerURLs)
	if err != nil {
		log.Log.Errorf("NewKingbusServer:startRaftPeer error,err:%s", err)
		return nil, err
	}
	err = s.startAdminServer(s.Cfg.AdminURLs)
	if err != nil {
		log.Log.Errorf("NewKingbusServer:startAdminServer error,err:%s", err)
		return nil, err
	}
	err = s.newBinlogProgress()
	if err != nil {
		log.Log.Errorf("NewKingbusServer:newBinlogProgress error,err:%s", err)
		return nil, err
	}
	err = s.startPrometheus(s.Cfg.MetricsAddr)
	if err != nil {
		log.Log.Errorf("NewKingbusServer:startPrometheus error,err:%s", err)
		return nil, err
	}
	s.started = atomic.NewBool(false)
	return s, nil
}
func (s *KingbusServer) updateLeadership(leadChange bool) {
	log.Log.Debugf("updateLeadership:leadChange:%v,s.IsLeader:%v,s.IsSyncerStarted:%v,",
		leadChange, s.IsLeader(), s.IsSyncerStarted())
	log.Log.Debugf("updateLeadership:appliedIndex:%d,committedIndex:%d", s.getAppliedIndex(), s.getCommittedIndex())
	if s.IsLeader() == false && s.IsSyncerStarted() {
		log.Log.Infof("server[%s] is not leader,stop syncer", s.Cfg.RaftNodeCfg.PeerURLs.String())
		s.StopServer(config.SyncerServerType)
		return
	}
	if leadChange && s.IsLeader() {
		go func() {
			t := time.Now()
			s.Mu.Lock()
			s.leadElectedTime = t
			s.Mu.Unlock()
			<-time.After(time.Second * 10)
			value, err := s.store.Get(utils.StringToBytes(storage.SyncerArgsKey))
			if err != nil {
				log.Log.Errorf("store Get value error,err:%s,key:%s,not start syncer",
					err, storage.SyncerArgsKey)
				return
			}
			if value == nil {
				log.Log.Infof("syncer didn't start, ignore start syncer")
				return
			}
			var syncerArgs config.SyncerArgs
			err = json.Unmarshal(value, &syncerArgs)
			if err != nil {
				log.Log.Fatalf("json unmarshal error,err:%s,value:%v", err, value)
			}
			err = s.StartServer(config.SyncerServerType, &syncerArgs)
			if err != nil {
				log.Log.Errorf("Start syncer error,err:%s,syncerAgs:%v", err, syncerArgs)
				return
			}
			log.Log.Infof("server[%s] become leader,start syncer automatically", s.Cfg.RaftNodeCfg.PeerURLs[0].Host)
		}()
	}
	if s.stats != nil {
		s.stats.BecomeLeader()
	}
}
func (s *KingbusServer) runRaft() {
	rh := &raft.ReadyHandler{
		GetLead:          func() (lead uint64) { return s.getLead() },
		UpdateLead:       func(lead uint64) { s.setLead(lead) },
		UpdateLeadership: s.updateLeadership,
		UpdateCommittedIndex: func(ci uint64) {
			cci := s.getCommittedIndex()
			if ci > cci {
				s.setCommittedIndex(ci)
			}
		},
	}
	s.raftNode.Run(rh)
	for {
		select {
		case ap := <-s.raftNode.Apply():
			s.applyAll(&ap)
		case <-s.raftNode.DoneC():
			log.Log.Infof("runRaft:quit success")
			return
		}
	}
}
func (s *KingbusServer) Run() {
	if s.started.Load() {
		return
	}
	go s.runRaft()
	go s.adminSvr.Run()
	go s.SyncAdminURL()
	go s.prometheusSvr.Run()
	s.started.Store(true)
	for {
		select {
		case err := <-s.errorc:
			log.Log.Errorf("KingbusServer.Start: run error,need quit. error:%s", err)
			s.Stop()
			return
		case <-s.stopping:
			log.Log.Infof("KingbusServer.Start quit")
			return
		}
	}
}