type RedisClent struct {
	// socket連接
	conn net.Conn
	addr string
	// 客戶端當前狀態(tài)
	connStatus atomic.Int32
	// heartbeat
	ticker time.Ticker
	// buffer cache
	waitSend   chan *request
	waitResult chan *request
	// 有請求正在處理中...
	working sync.WaitGroup
}
// 創(chuàng)建redis客戶端socket
func NewRedisClient(addr string) (*RedisClent, error) {
	conn, err := net.Dial("tcp", addr)
	if err != nil {
		returnnil, err
	}
	rc := RedisClent{}
	rc.conn = conn
	rc.waitSend = make(chan *request, maxChanSize)
	rc.waitResult = make(chan *request, maxChanSize)
	rc.addr = addr
	return &rc, nil
}
// 啟動
func (rc *RedisClent) Start() error {
	rc.ticker = *time.NewTicker(heartBeatInterval)
	// 將waitSend緩沖區(qū)進行發(fā)送
	go rc.execSend()
	// 獲取服務端結果
	go rc.execReceive()
	// 定時發(fā)送心跳
	go rc.execHeardBeat()
	rc.connStatus.Store(connRunning) // 啟動狀態(tài)
	returnnil
}

// 將redis命令保存到 waitSend 中
func (rc *RedisClent) Send(command [][]byte) (protocol.Reply, error) {
	// 已關閉
	if rc.connStatus.Load() == connClosed {
		returnnil, errors.New("client closed")
	}
	req := &request{
		command: command,
		wait:    wait.Wait{},
	}
	// 單個請求
	req.wait.Add(1)
	// 所有請求
	rc.working.Add(1)
	defer rc.working.Done()
	// 將數(shù)據(jù)保存到緩沖中
	rc.waitSend <- req
	// 等待處理結束
	if req.wait.WaitWithTimeOut(maxWait) {
		returnnil, errors.New("time out")
	}
	// 出錯
	if req.err != nil {
		err := req.err
		returnnil, err
	}
	// 正常
	return req.reply, nil
}

// 將waitSend緩沖區(qū)進行發(fā)送
func (rc *RedisClent) execSend() {
	for req := range rc.waitSend {
		rc.sendReq(req)
	}
}
func (rc *RedisClent) sendReq(req *request) {
	// 無效請求
	if req == nil || len(req.command) == 0 {
		return
	}
	var err error
	// 網(wǎng)絡請求（重試3次）
	for i := 0; i < 3; i++ {
		_, err = rc.conn.Write(req.Bytes())
		// 發(fā)送成功 or 發(fā)送錯誤（除了超時錯誤和deadline錯誤）跳出
		if err == nil ||
			(!strings.Contains(err.Error(), "timeout") && // only retry timeout
				!strings.Contains(err.Error(), "deadline exceeded")) {
			break
		}
	}
	if err == nil { // 發(fā)送成功，異步等待結果
		rc.waitResult <- req
	} else { // 發(fā)送失敗，請求直接失敗
		req.err = err
		req.wait.Done()
	}
}

func (rc *RedisClent) execReceive() {
	ch := parser.ParseStream(rc.conn)
	for payload := range ch {
		if payload.Err != nil {
			if rc.connStatus.Load() == connClosed { // 連接已關閉
				return
			}
			// 否則，重新連接（可能因為網(wǎng)絡抖動臨時斷開了）
			rc.reconnect()
			return
		}
		// 說明一切正常
		rc.handleResult(payload.Reply)
	}
}
func (rc *RedisClent) handleResult(reply protocol.Reply) {
	// 從rc.waitResult 獲取一個等待中的請求，將結果保存進去
	req := <-rc.waitResult
	if req == nil {
		return
	}
	req.reply = reply
	req.wait.Done() // 通知已經(jīng)獲取到結果
}

func (rc *RedisClent) reconnect() {
	logger.Info("redis client reconnect...")
	rc.conn.Close()
	var conn net.Conn
	// 重連（重試3次）
	for i := 0; i < 3; i++ {
		var err error
		conn, err = net.Dial("tcp", rc.addr)
		if err != nil {
			logger.Error("reconnect error: " + err.Error())
			time.Sleep(time.Second)
			continue
		} else {
			break
		}
	}
	// 服務端連不上，說明服務可能掛了（or 網(wǎng)絡問題 and so on...)
	if conn == nil { 
		rc.Stop()
		return
	}
	// 這里關閉沒問題，因為rc.conn.Close已經(jīng)關閉，函數(shù)Send中保存的請求因為發(fā)送不成功，不會寫入到waitResult
	close(rc.waitResult)
	// 清理 waitResult(因為連接重置，新連接上只能處理新請求，老的請求的數(shù)據(jù)結果在老連接上,老連接已經(jīng)關了，新連接上肯定是沒有結果的)
	for req := range rc.waitResult {
		req.err = errors.New("connect reset")
		req.wait.Done()
	}
	// 新連接（新氣象）
	rc.waitResult = make(chan *request, maxWait)
	rc.conn = conn
	// 重新啟動接收協(xié)程
	go rc.execReceive()
}