Golang实现Redis网络协议实例探究

func (t *TCPServer) handleConn(conn net.Conn) {
	// ...代码省略...
	
	logger.Debugf("accept new conn %s", conn.RemoteAddr().String())

	// TODO :处理连接
	t.redisHander.Handle(context.Background(), conn)
}

// 该方法是不同的conn复用的方法，要做的事情就是从conn中读取出符合RESP格式的数据；
// 然后针对消息格式，进行不同的业务处理
func (h *RedisHandler) Handle(ctx context.Context, conn net.Conn) {
	h.activeConn.Store(conn, struct{}{})
	outChan := parser.ParseStream(conn)
	for payload := range outChan {
		if payload.Err != nil {
			// 网络conn关闭
			if payload.Err == io.EOF || payload.Err == io.ErrUnexpectedEOF || strings.Contains(payload.Err.Error(), "use of closed network connection") {
				h.activeConn.Delete(conn)
				conn.Close()
				logger.Warn("client closed:" + conn.RemoteAddr().String())
				return
			}
			// 解析出错 protocol error
			errReply := protocal.NewGenericErrReply(payload.Err.Error())
			_, err := conn.Write(errReply.ToBytes())
			if err != nil {
				h.activeConn.Delete(conn)
				conn.Close()
				logger.Warn("client closed:" + conn.RemoteAddr().String() + " err info: " + err.Error())
				return
			}
			continue
		}
		if payload.Reply == nil {
			logger.Error("empty payload")
			continue
		}
		reply, ok := payload.Reply.(*protocal.MultiBulkReply)
		if !ok {
			logger.Error("require multi bulk protocol")
			continue
		}
		logger.Debugf("%q", string(reply.ToBytes()))
		result := h.engine.Exec(conn, reply.RedisCommand)
		if result != nil {
			conn.Write(result.ToBytes())
		} else {
			conn.Write(protocal.NewUnknownErrReply().ToBytes())
		}
	}
}

// 从r中读取数据，将读取的结果通过 out chan 发送给外部使用（包括：正常的数据包 or 网络错误）
func parse(r io.Reader, out chan<- *Payload) {
	// 异常恢复，避免未知异常
	deferfunc() {
		if err := recover(); err != nil {
			logger.Error(err, string(debug.Stack()))
		}
	}()
	reader := bufio.NewReader(r)
	for {
		// 按照 \n 分隔符读取一行数据
		line, err := reader.ReadBytes('\n')
		if err != nil { // 一般是 io.EOF错误（说明conn关闭or文件尾部）
			out <- &Payload{Err: err}
			close(out)
			return
		}
		// 读取到的line中包括 \n 分割符
		length := len(line)
		// RESP协议是按照 \r\n 分割数据
		if length <= 2 || line[length-2] != '\r' { // 说明是空白行，忽略
			continue
		}
		// 去掉尾部 \r\n
		line = bytes.TrimSuffix(line, []byte{'\r', '\n'})
		// 协议文档 ：https://redis.io/docs/reference/protocol-spec/
		// The first byte in an RESP-serialized payload always identifies its type. Subsequent bytes constitute the type's contents.
		switch line[0] {
		case'*': // * 表示数组
			err := parseArrays(line, reader, out)
			if err != nil {
				out <- &Payload{Err: err}
				close(out)
				return
			}
		default:
			args := bytes.Split(line, []byte{' '})
			out <- &Payload{
				Reply: protocal.NewMultiBulkReply(args),
			}
		}
	}
}

// 基于数字5 读取 5+2 长度的数据，这里的2表示\r\n
body := make([]byte, dataLen+2)
// 注意：这里直接读取指定长度的字节
_, err := io.ReadFull(reader, body)
if err != nil {
    return err
}
// 所以最终读取到的是 hello\r\n，去掉\r\n 保存到 lines中
lines = append(lines, body[:len(body)-2])

*2\r\n$3\r\nget\r\n$9\r\neasyredis\r\n
*3\r\n$3\r\nset\r\n$9\r\neasyredis\r\n$1\r\n1\r\n

func NewReader(rd io.Reader) *Reader

file, err := os.Open("example.txt")
if err != nil {
    log.Fatal(err)
}
defer file.Close()
reader := bufio.NewReader(file)

func NewScanner(r io.Reader) *Scanner

file, err := os.Open("example.txt")
if err != nil {
    log.Fatal(err)
}
defer file.Close()

scanner := bufio.NewScanner(file)