基于Golang如何实现Redis协议解析器
Redis Assistant 是一款为Redis设计的高效可视化管理与监控工具，提供多种操作模式，高效处理服务器数据。日常使用里的持续防护和信息管理会更突出，适合把安全控制放进长期使用流程中的场景。
RESP协议RESP是客户端与服务端通信的协议，格式有五种：正常回复：以“+”开头，以“\r\n”结尾的字符串形式错误回复：以“-”开头，以“\r\n”结尾的字符串形式整数：以“:”开头，以“\r\n”结尾的字符串形式多行字符串：以“$”开头，后跟实际发送字节数，再以“\r\n”开头和结尾$3\r\nabc\r\n数组：以“*”开头，后跟成员个数SETkeyvalue*3\r\n$3\r\nSET\r\n$3\r\nkey\r\n$5\r\nvalue\r\n客户端和服务器发送的命令或数据一律以\r\n（
RESP协议
RESP是客户端与服务端通信的协议，格式有五种：
正常回复：以“+”开头，以“\r\n”结尾的字符串形式
错误回复：以“-”开头，以“\r\n”结尾的字符串形式
整数：以“:”开头，以“\r\n”结尾的字符串形式
多行字符串：以“$”开头，后跟实际发送字节数，再以“\r\n”开头和结尾
$3\r\nabc\r\n
数组：以“*”开头，后跟成员个数
SET key value
*3\r\n$3\r\nSET\r\n$3\r\nkey\r\n$5\r\nvalue\r\n
客户端和服务器发送的命令或数据一律以 \r\n （CRLF）作为换行符。
当我们输入*3\r\n$3\r\nSET\r\n$3\r\nkey\r\n$5\r\nvalue\r\n这样一串命令，服务端接收到的是如下的命令：
*3\r\n
$3\r\n
SET\r\n
$3\r\n
key\r\n
$5\r\n
value\r\n
interface/resp/conn.go
type Connection interface {
Write([]byte) error
GetDBIndex() int
SelectDB(int)
}
interface/resp/reply.go
type Reply interface {
ToBytes() []byte
}Connection接口：Redis客户端的一个连接
Write：给客户端回复消息
GetDBIndex：Redis有16个DB
Reply接口：响应接口
resp/reply/consts.go
type PongReply struct{}
var pongBytes = []byte("+PONG\r\n")
func (r *PongReply) ToBytes() []byte {
return pongBytes
}
var thePongReply = new(PongReply)
func MakePongReply() *PongReply {
return thePongReply
}
type OkReply struct{}
var okBytes = []byte("+OK\r\n")
func (r *OkReply) ToBytes() []byte {
return okBytes
}
var theOkReply = new(OkReply)
func MakeOkReply() *OkReply {
return theOkReply
}
var nullBulkBytes = []byte("$-1\r\n")
type NullBulkReply struct{}
func (r *NullBulkReply) ToBytes() []byte {
return nullBulkBytes
}
func MakeNullBulkReply() *NullBulkReply {
return &NullBulkReply{}
}
var emptyMultiBulkBytes = []byte("*0\r\n")
type EmptyMultiBulkReply struct{}
func (r *EmptyMultiBulkReply) ToBytes() []byte {
return emptyMultiBulkBytes
}
type NoReply struct{}
var noBytes = []byte("")
func (r *NoReply) ToBytes() []byte {
return noBytes
}定义五种回复：回复pong，ok，null，空数组，空
resp/reply/reply.go
type ErrorReply interface {
Error() string
ToBytes() []byte
}ErrorReply：定义错误接口
resp/reply/errors.go
type UnknownErrReply struct{}
var unknownErrBytes = []byte("-Err unknown\r\n")
func (r *UnknownErrReply) ToBytes() []byte {
return unknownErrBytes
}
func (r *UnknownErrReply) Error() string {
return "Err unknown"
}
type ArgNumErrReply struct {
Cmd string
}
func (r *ArgNumErrReply) ToBytes() []byte {
return []byte("-ERR wrong number of arguments for '" + r.Cmd + "' command\r\n")
}
func (r *ArgNumErrReply) Error() string {
return "ERR wrong number of arguments for '" + r.Cmd + "' command"
}
func MakeArgNumErrReply(cmd string) *ArgNumErrReply {
return &ArgNumErrReply{
Cmd: cmd,
}
}
type SyntaxErrReply struct{}
var syntaxErrBytes = []byte("-Err syntax error\r\n")
var theSyntaxErrReply = &SyntaxErrReply{}
func MakeSyntaxErrReply() *SyntaxErrReply {
return theSyntaxErrReply
}
func (r *SyntaxErrReply) ToBytes() []byte {
return syntaxErrBytes
}
func (r *SyntaxErrReply) Error() string {
return "Err syntax error"
}
type WrongTypeErrReply struct{}
var wrongTypeErrBytes = []byte("-WRONGTYPE Operation against a key holding the wrong kind of value\r\n")
func (r *WrongTypeErrReply) ToBytes() []byte {
return wrongTypeErrBytes
}
func (r *WrongTypeErrReply) Error() string {
return "WRONGTYPE Operation against a key holding the wrong kind of value"
}
type ProtocolErrReply struct {
Msg string
}
func (r *ProtocolErrReply) ToBytes() []byte {
return []byte("-ERR Protocol error: '" + r.Msg + "'\r\n")
}
func (r *ProtocolErrReply) Error() string {
return "ERR Protocol error: '" + r.Msg
}errors定义5种错误：UnknownErrReply 未知错误，ArgNumErrReply 参数个数错误，SyntaxErrReply 语法错误，WrongTypeErrReply 数据类型错误，ProtocolErrReply 协议错误
resp/reply/reply.go
var (
nullBulkReplyBytes = []byte("$-1")
// 协议的结尾
CRLF = "\r\n"
)
type BulkReply struct {
Arg []byte
}
func MakeBulkReply(arg []byte) *BulkReply {
return &BulkReply{
Arg: arg,
}
}
func (r *BulkReply) ToBytes() []byte {
if len(r.Arg) == 0 {
return nullBulkReplyBytes
}
return []byte("$" + strconv.Itoa(len(r.Arg)) + CRLF + string(r.Arg) + CRLF)
}
type MultiBulkReply struct {
Args [][]byte
}
func (r *MultiBulkReply) ToBytes() []byte {
argLen := len(r.Args)
var buf bytes.Buffer
buf.WriteString("*" + strconv.Itoa(argLen) + CRLF)
for _, arg := range r.Args {
if arg == nil {
buf.WriteString("$-1" + CRLF)
} else {
buf.WriteString("$" + strconv.Itoa(len(arg)) + CRLF + string(arg) + CRLF)
}
}
return buf.Bytes()
}
func MakeMultiBulkReply(args [][]byte) *MultiBulkReply {
return &MultiBulkReply{
Args: args,
}
}
type StatusReply struct {
Status string
}
func MakeStatusReply(status string) *StatusReply {
return &StatusReply{
Status: status,
}
}
func (r *StatusReply) ToBytes() []byte {
return []byte("+" + r.Status + CRLF)
}
type IntReply struct {
Code int64
}
func MakeIntReply(code int64) *IntReply {
return &IntReply{
Code: code,
}
}
func (r *IntReply) ToBytes() []byte {
return []byte(":" + strconv.FormatInt(r.Code, 10) + CRLF)
}
type StandardErrReply struct {
Status string
}
func (r *StandardErrReply) ToBytes() []byte {
return []byte("-" + r.Status + CRLF)
}
func (r *StandardErrReply) Error() string {
return r.Status
}
func MakeErrReply(status string) *StandardErrReply {
return &StandardErrReply{
Status: status,
}
}
func IsErrorReply(reply resp.Reply) bool {
return reply.ToBytes()[0] == '-'
}BulkReply：回复一个字符串
MultiBulkReply：回复字符串数组
StatusReply：状态回复
IntReply：数字回复
StandardErrReply：标准错误回复
IsErrorReply：判断是否为错误回复
ToBytes：将字符串转成RESP协议规定的格式
resp/parser/parser.go
type Payload struct {
Data resp.Reply
Err error
}
type readState struct {
readingMultiLine bool
expectedArgsCount int
msgType byte
args [][]byte
bulkLen int64
}
func (s *readState) finished() bool {
return s.expectedArgsCount > 0 && len(s.args) == s.expectedArgsCount
}
func ParseStream(reader io.Reader) <-chan *Payload {
ch := make(chan *Payload)
go parse0(reader, ch)
return ch
}
func parse0(reader io.Reader, ch chan<- *Payload) {
......
}Payload结构体：客服端给我们发的数据
Reply：客户端与服务端互相发的数据都称为Reply
readState结构体：
readingMultiLine：解析单行还是多行数据
expectedArgsCount：应该读取的参数个数
msgType：消息类型
args：消息内容
bulkLen：数据长度
finished方法：判断解析是否完成
ParseStream方法：异步解析数据后放入管道，返回管道数据
func readLine(bufReader *bufio.Reader, state *readState) ([]byte, bool, error) {
var msg []byte
var err error
if state.bulkLen == 0 {
msg, err = bufReader.ReadBytes('\n')
if err != nil {
return nil, true, err
}
if len(msg) == 0 || msg[len(msg)-2] != '\r' {
return nil, false, errors.New("protocol error: " + string(msg))
}
} else {
msg = make([]byte, state.bulkLen+2)
_, err = io.ReadFull(bufReader, msg)
if err != nil {
return nil, true, err
}
if len(msg) == 0 || msg[len(msg)-2] != '\r' || msg[len(msg)-1] != '\n' {
return nil, false, errors.New("protocol error: " + string(msg))
}
state.bulkLen = 0
}
return msg, false, nil
}readLine：一行一行的读取。读正常的行，以\n分隔。读正文中包含\r\n字符的行时，state.bulkLen加上换行符\r\n（state.bulkLen+2）
func parseMultiBulkHeader(msg []byte, state *readState) error {
var err error
var expectedLine uint64
expectedLine, err = strconv.ParseUint(string(msg[1:len(msg)-2]), 10, 32)
if err != nil {
return errors.New("protocol error: " + string(msg))
}
if expectedLine == 0 {
state.expectedArgsCount = 0
return nil
} else if expectedLine > 0 {
state.msgType = msg[0]
state.readingMultiLine = true
state.expectedArgsCount = int(expectedLine)
state.args = make([][]byte, 0, expectedLine)
return nil
} else {
return errors.New("protocol error: " + string(msg))
}
}
func parseBulkHeader(msg []byte, state *readState) error {
var err error
state.bulkLen, err = strconv.ParseInt(string(msg[1:len(msg)-2]), 10, 64)
if err != nil {
return errors.New("protocol error: " + string(msg))
}
if state.bulkLen == -1 { // null bulk
return nil
} else if state.bulkLen > 0 {
state.msgType = msg[0]
state.readingMultiLine = true
state.expectedArgsCount = 1
state.args = make([][]byte, 0, 1)
return nil
} else {
return errors.New("protocol error: " + string(msg))
}
}parseMultiBulkHeader：解析数组的头部，设置期望的行数和相关参数。
parseBulkHeader：解析多行字符串的头部。
func parseSingleLineReply(msg []byte) (resp.Reply, error) {
str := strings.TrimSuffix(string(msg), "\r\n")
var result resp.Reply
switch msg[0] {
case '+': // status reply
result = reply.MakeStatusReply(str[1:])
case '-': // err reply
result = reply.MakeErrReply(str[1:])
case ':': // int reply
val, err := strconv.ParseInt(str[1:], 10, 64)
if err != nil {
return nil, errors.New("protocol error: " + string(msg))
}
result = reply.MakeIntReply(val)
}
return result, nil
}
func readBody(msg []byte, state *readState) error {
line := msg[0 : len(msg)-2]
var err error
if line[0] == '$' {
// bulk reply
state.bulkLen, err = strconv.ParseInt(string(line[1:]), 10, 64)
if err != nil {
return errors.New("protocol error: " + string(msg))
}
if state.bulkLen <= 0 { // null bulk in multi bulks
state.args = append(state.args, []byte{})
state.bulkLen = 0
}
} else {
state.args = append(state.args, line)
}
return nil
}parseSingleLineReply：解析单行命令
readBody：读取多行的命令，如果是$开头，设置bulkLen，读取下一行时根据这个+2，不是$开头则直接添加到args
func parse0(reader io.Reader, ch chan<- *Payload) {
defer func() {
if err := recover(); err != nil {
logger.Error(string(debug.Stack()))
}
}()
bufReader := bufio.NewReader(reader)
var state readState
var err error
var msg []byte
for {
var ioErr bool
msg, ioErr, err = readLine(bufReader, &state)
if err != nil {
if ioErr {
ch <- &Payload{
Err: err,
}
close(ch)
return
}
ch <- &Payload{
Err: err,
}
state = readState{}
continue
}
if !state.readingMultiLine {
if msg[0] == '*' {
// multi bulk reply
err = parseMultiBulkHeader(msg, &state)
if err != nil {
ch <- &Payload{
Err: errors.New("protocol error: " + string(msg)),
}
state = readState{}
continue
}
if state.expectedArgsCount == 0 {
ch <- &Payload{
Data: &reply.EmptyMultiBulkReply{},
}
state = readState{}
continue
}
} else if msg[0] == '$' { // bulk reply
err = parseBulkHeader(msg, &state)
if err != nil {
ch <- &Payload{
Err: errors.New("protocol error: " + string(msg)),
}
state = readState{} // reset state
continue
}
if state.bulkLen == -1 { // null bulk reply
ch <- &Payload{
Data: &reply.NullBulkReply{},
}
state = readState{} // reset state
continue
}
} else {
// single line reply
result, err := parseSingleLineReply(msg)
ch <- &Payload{
Data: result,
Err: err,
}
state = readState{} // reset state
continue
}
} else {
// read bulk reply
err = readBody(msg, &state)
if err != nil {
ch <- &Payload{
Err: errors.New("protocol error: " + string(msg)),
}
state = readState{} // reset state
continue
}
// if sending finished
if state.finished() {
var result resp.Reply
if state.msgType == '*' {
result = reply.MakeMultiBulkReply(state.args)
} else if state.msgType == '$' {
result = reply.MakeBulkReply(state.args[0])
}
ch <- &Payload{
Data: result,
Err: err,
}
state = readState{}
}
}
}
}parse0：解析指令，解析完成后通过channel发出去
resp/connection/conn.go
type Connection struct {
conn net.Conn
waitingReply wait.Wait
mu sync.Mutex // 避免多个协程往客户端中写
selectedDB int
}
func NewConn(conn net.Conn) *Connection {
return &Connection{
conn: conn,
}
}
func (c *Connection) RemoteAddr() net.Addr {
return c.conn.RemoteAddr()
}
func (c *Connection) Close() error {
c.waitingReply.WaitWithTimeout(10 * time.Second)
_ = c.conn.Close()
return nil
}
func (c *Connection) Write(b []byte) error {
if len(b) == 0 {
return nil
}
c.mu.Lock()
c.waitingReply.Add(1)
defer func() {
c.waitingReply.Done()
c.mu.Unlock()
}()
_, err := c.conn.Write(b)
return err
}
func (c *Connection) GetDBIndex() int {
return c.selectedDB
}
func (c *Connection) SelectDB(dbNum int) {
c.selectedDB = dbNum
}之前写的EchoHandler是用户传过来什么，我们传回去什么。现在要写一个RespHandler来代替EchoHandler，让解析器来解析。RespHandler中要有一个管理客户端连接的结构体Connection。
Connection：客户端连接，在协议层的handler中会用到
resp/handler/handler.go
var (
unknownErrReplyBytes = []byte("-ERR unknown\r\n")
)
type RespHandler struct {
activeConn sync.Map
db databaseface.Database
closing atomic.Boolean
}
func MakeHandler() *RespHandler {
var db databaseface.Database
db = database.NewEchoDatabase()
return &RespHandler{
db: db,
}
}
func (h *RespHandler) closeClient(client *connection.Connection) {
_ = client.Close()
h.db.AfterClientClose(client)
h.activeConn.Delete(client)
}
func (h *RespHandler) Handle(ctx context.Context, conn net.Conn) {
if h.closing.Get() {
// closing handler refuse new connection
_ = conn.Close()
}
client := connection.NewConn(conn)
h.activeConn.Store(client, 1)
ch := parser.ParseStream(conn)
for payload := range ch {
if payload.Err != nil {
if payload.Err == io.EOF ||
payload.Err == io.ErrUnexpectedEOF ||
strings.Contains(payload.Err.Error(), "use of closed network connection") {
// connection closed
h.closeClient(client)
logger.Info("connection closed: " + client.RemoteAddr().String())
return
}
// protocol err
errReply := reply.MakeErrReply(payload.Err.Error())
err := client.Write(errReply.ToBytes())
if err != nil {
h.closeClient(client)
logger.Info("connection closed: " + client.RemoteAddr().String())
return
}
continue
}
if payload.Data == nil {
logger.Error("empty payload")
continue
}
r, ok := payload.Data.(*reply.MultiBulkReply)
if !ok {
logger.Error("require multi bulk reply")
continue
}
result := h.db.Exec(client, r.Args)
if result != nil {
_ = client.Write(result.ToBytes())
} else {
_ = client.Write(unknownErrReplyBytes)
}
}
}
func (h *RespHandler) Close() error {
logger.Info("handler shutting down...")
h.closing.Set(true)
// TODO: concurrent wait
h.activeConn.Range(func(key interface{}, val interface{}) bool {
client := key.(*connection.Connection)
_ = client.Close()
return true
})
h.db.Close()
return nil
}RespHandler：和之前的echo类似，加了核心层的db.exec执行解析的指令
interface/database/database.go
type CmdLine = [][]byte
type Database interface {
Exec(client resp.Connection, args [][]byte) resp.Reply
AfterClientClose(c resp.Connection)
Close()
}
type DataEntity struct {
Data interface{}
}Exec：核心层的执行
AfterClientClose：关闭之后的善后方法
CmdLine：二维字节数组的指令别名
DataEntity：表示Redis的数据，包括string, list, set等等
database/echo_database.go
type EchoDatabase struct {
}
func NewEchoDatabase() *EchoDatabase {
return &EchoDatabase{}
}
func (e EchoDatabase) Exec(client resp.Connection, args [][]byte) resp.Reply {
return reply.MakeMultiBulkReply(args)
}
func (e EchoDatabase) AfterClientClose(c resp.Connection) {
logger.Info("EchoDatabase AfterClientClose")
}
func (e EchoDatabase) Close() {
logger.Info("EchoDatabase Close")
}echo_database：测试协议层
Exec：指令解析后，再使用MakeMultiBulkReply包装一下返回去
main.go
err := tcp.ListenAndServeWithSignal(
&tcp.Config{
Address: fmt.Sprintf("%s:%d",
config.Properties.Bind,
config.Properties.Port),
},
handler.MakeHandler())
if err != nil {
logger.Error(err)
}main改成刚才写的：handler.MakeHandler()
Postman是一套面向API全生命周期的开发与测试平台，可在Mac上发送和调试多种协议请求，管理集合、环境及认证信息，并完成脚本测试、模拟服务、监控、文档发布和团队协作。它适合开发、测试及运维人员使用，提供个人免费方案。
Docker Desktop 是 Docker 官方为 Mac 提供的容器开发环境，集成 Docker Engine、命令行工具、Compose、Build、Kubernetes 与图形化管理界面。
Android Studio 是 Google 面向安卓应用开发提供的官方集成开发环境，基于 IntelliJ IDEA 构建。它整合智能代码编辑、Gradle 构建、界面预览、设备模拟、性能分析、调试测试及 APK 分析等工具。
Xcode 是苹果面向 Mac 提供的官方集成开发环境，将代码编辑、项目构建、SwiftUI 实时预览、设备模拟、自动化测试、断点调试、性能分析和应用分发整合在同一套工作流中，适合开发与维护 iPhone、iPad、Mac、Apple Wa
JetBrains WebStorm 是面向 JavaScript、TypeScript 和现代 Web 技术的专业集成开发环境，可智能理解项目结构，提供代码补全、重构、错误检查、运行调试、测试及 Git 工具。
PhpStorm 是 JetBrains 推出的专业 PHP 集成开发环境，可在 Mac 上完成代码编写、智能检查、重构、调试、测试和数据库管理。它支持主流 PHP 框架、Composer、Git、Docker 与远程解释器。
JetBrains PyCharm 是面向 Python 开发者的集成开发环境，提供智能补全、代码检查、重构、调试、测试、版本控制及虚拟环境管理功能。Mac 版兼容 Intel 与 Apple 芯片，核心功能可免费使用。
IntelliJ IDEA 是 JetBrains 推出的专业集成开发环境，重点服务于 Java、Kotlin 与 JVM 项目开发。软件集成代码补全、静态检查、重构、调试、测试、版本控制及构建工具，并可借助插件扩展语言和框架支持。
Visual Studio Code 是微软提供的免费代码编辑器，支持 macOS，并为 Intel 与 Apple 芯片 Mac 提供适配版本。软件集代码编辑、调试、Git 源代码管理、集成终端和扩展市场于一体。
REAPER是Cockos开发的数字音频工作站，提供多轨音频与MIDI录制、剪辑、处理、混音和母带制作工具。Mac版兼容Intel与Apple芯片，支持AU、VST、VST3、CLAP等插件格式，并提供高度可定制的工作流程。
Adobe Audition 是面向播客、影视后期、音乐制作及内容创作者的专业音频工作站，提供波形精修、多轨录制与混音、频谱编辑、降噪修复、响度匹配和效果处理工具，并可与 Adobe Premiere 协同完成视频声音制作。
Pro Tools 是 Avid 推出的专业数字音频工作站，覆盖多轨录音、无损剪辑、MIDI 编曲、插件处理、自动化混音及声音与画面同步等工作。它拥有 Intro、Artist、Studio 和 Ultimate 等方案。
Postman 是面向 API 全生命周期的开发与协作平台。Windows 桌面版可用于构建和发送 HTTP、GraphQL、gRPC、WebSocket、MQTT 等请求，并提供环境变量、集合、脚本测试、模拟服务、监控、文档及团队工作区等
Docker Desktop 为 Windows 开发者提供集成式容器环境，内置 Docker Engine、命令行工具、Compose、Build 与 Kubernetes 等组件，可在本机完成镜像构建、容器运行、日志查看和资源管理。
Android Studio 是 Google 面向 Android 应用开发提供的官方集成开发环境，以 IntelliJ IDEA 为基础，集成智能代码编辑、Gradle 构建、设备模拟、界面预览、调试测试及性能分析工具，
WebStorm 是 JetBrains 面向 JavaScript、TypeScript 和现代 Web 技术打造的集成开发环境。它提供项目级代码分析、智能补全、安全重构、导航搜索，并集成调试、测试、终端、Git 与常用前端工具，
PhpStorm 是 JetBrains 面向 PHP 与现代 Web 开发打造的专业集成开发环境，提供代码补全、静态检查、重构、调试、测试、数据库、版本控制及前端语言支持。Windows 用户可直接下载安装程序，
PyCharm 是 JetBrains 推出的 Python 集成开发环境，提供智能代码补全、错误检查、重构、调试、测试、Git、终端及 Jupyter Notebook 支持。统一版安装后可体验 30 天 Pro 功能，
JetBrains IntelliJ IDEA 是以 Java、Kotlin 和 JVM 项目为核心的专业集成开发环境，提供智能补全、代码检查、安全重构、调试、测试、版本控制及构建工具集成。统一发行版的核心功能可免费用于商业和非商业项目，
Visual Studio Code 是微软推出的免费代码编辑器，Windows 版集成智能补全、代码导航、调试、终端和版本控制，并可通过扩展添加语言、格式化及远程开发能力。它启动较快、配置灵活，
REAPER 是 Cockos 推出的数字音频工作站，适用于录音、MIDI 编曲、音频编辑、混音、母带制作及声音设计。软件安装包小巧，支持多轨录制、灵活路由、自动化、第三方插件和脚本扩展，并提供完整功能的 60 天免费评估期。
Adobe Audition 是面向播客、影视后期、广播与音乐制作的专业音频工作站，提供波形编辑、多轨混音、频谱分析、录音、降噪修复和效果处理等功能。Windows 版通过 Adobe Creative Cloud 安装和管理，
Pro Tools 是 Avid 推出的专业数字音频工作站，提供多轨录音、音频剪辑、MIDI编曲、插件处理、自动化混音及影视声音制作能力。它适合音乐人、录音师、制作人和音频后期团队使用，并提供免费入门版、订阅方案及永久授权选择。
Cubase 是 Steinberg 推出的数字音频工作站，提供 MIDI 编曲、多轨录音、音频编辑、虚拟乐器、效果处理、自动化、混音和母带输出工具。其 Pro、Artist 与 Elements 版本面向不同预算和制作规模，