type RedisClent struct {
	// socket連接
	conn net.Conn
	addr string
	// 客戶端當(dāng)前狀態(tài)
	connStatus atomic.Int32
	// heartbeat
	ticker time.Ticker
	// buffer cache
	waitSend   chan *request
	waitResult chan *request
	// 有請(qǐng)求正在處理中...
	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
}
// 啟動(dòng)
func (rc *RedisClent) Start() error {
	rc.ticker = *time.NewTicker(heartBeatInterval)
	// 將waitSend緩沖區(qū)進(jìn)行發(fā)送
	go rc.execSend()
	// 獲取服務(wù)端結(jié)果
	go rc.execReceive()
	// 定時(shí)發(fā)送心跳
	go rc.execHeardBeat()
	rc.connStatus.Store(connRunning) // 啟動(dòng)狀態(tài)
	returnnil
}

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

// 將waitSend緩沖區(qū)進(jìn)行發(fā)送
func (rc *RedisClent) execSend() {
	for req := range rc.waitSend {
		rc.sendReq(req)
	}
}
func (rc *RedisClent) sendReq(req *request) {
	// 無(wú)效請(qǐng)求
	if req == nil || len(req.command) == 0 {
		return
	}
	var err error
	// 網(wǎng)絡(luò)請(qǐng)求（重試3次）
	for i := 0; i < 3; i++ {
		_, err = rc.conn.Write(req.Bytes())
		// 發(fā)送成功 or 發(fā)送錯(cuò)誤（除了超時(shí)錯(cuò)誤和deadline錯(cuò)誤）跳出
		if err == nil ||
			(!strings.Contains(err.Error(), "timeout") && // only retry timeout
				!strings.Contains(err.Error(), "deadline exceeded")) {
			break
		}
	}
	if err == nil { // 發(fā)送成功，異步等待結(jié)果
		rc.waitResult <- req
	} else { // 發(fā)送失敗，請(qǐng)求直接失敗
		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 { // 連接已關(guān)閉
				return
			}
			// 否則，重新連接（可能因?yàn)榫W(wǎng)絡(luò)抖動(dòng)臨時(shí)斷開了）
			rc.reconnect()
			return
		}
		// 說(shuō)明一切正常
		rc.handleResult(payload.Reply)
	}
}
func (rc *RedisClent) handleResult(reply protocol.Reply) {
	// 從rc.waitResult 獲取一個(gè)等待中的請(qǐng)求，將結(jié)果保存進(jìn)去
	req := <-rc.waitResult
	if req == nil {
		return
	}
	req.reply = reply
	req.wait.Done() // 通知已經(jīng)獲取到結(jié)果
}

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
		}
	}
	// 服務(wù)端連不上，說(shuō)明服務(wù)可能掛了（or 網(wǎng)絡(luò)問(wèn)題 and so on...)
	if conn == nil { 
		rc.Stop()
		return
	}
	// 這里關(guān)閉沒(méi)問(wèn)題，因?yàn)閞c.conn.Close已經(jīng)關(guān)閉，函數(shù)Send中保存的請(qǐng)求因?yàn)榘l(fā)送不成功，不會(huì)寫入到waitResult
	close(rc.waitResult)
	// 清理 waitResult(因?yàn)檫B接重置，新連接上只能處理新請(qǐng)求，老的請(qǐng)求的數(shù)據(jù)結(jié)果在老連接上,老連接已經(jīng)關(guān)了，新連接上肯定是沒(méi)有結(jié)果的)
	for req := range rc.waitResult {
		req.err = errors.New("connect reset")
		req.wait.Done()
	}
	// 新連接（新氣象）
	rc.waitResult = make(chan *request, maxWait)
	rc.conn = conn
	// 重新啟動(dòng)接收協(xié)程
	go rc.execReceive()
}