Golang을 기반으로 Redis 프로토콜 파서를 구현하는 방법
- WBOY앞으로
- 2023-05-28 19:13:21753검색
RESP 프로토콜
RESP는 클라이언트와 서버 간의 통신을 위한 프로토콜입니다. 5가지 형식이 있습니다.
일반 응답: "+"로 시작하고 "rn"으로 끝나는 문자열
오류 응답: 다음으로 시작합니다. " -"는 "rn"으로 시작하고 "rn"으로 끝나는 문자열입니다.
정수: ":"으로 시작하고 "rn"으로 끝나는 문자열
여러 줄 문자열: "$"로 시작하고 실제 문자가 옵니다. 전송된 바이트 수("rn"
$3rnabcrn
배열: "*"로 시작하고 그 뒤에 멤버 수
SET 키 값
*3rn$3rnSET가 옴) rn$3 rnkeyrn$5rnvaluern
클라이언트와 서버에서 보낸 명령이나 데이터는 rn(CRLF)을 줄바꿈 문자로 사용해야 합니다.
*3rn$3rnSETrn$3rnkeyrn$5rnvaluern과 같은 일련의 명령을 입력하면 서버는 다음 명령을 받습니다:
*3rn
$3rn
SETrn
$3rn
keyrn
$5rn
valuern
interface/resp/conn.go
type Connection interface {
Write([]byte) error
GetDBIndex() int
SelectDB(int)
}
interface/resp/reply.go
type Reply interface {
ToBytes() []byte
}연결 인터페이스: Redis 클라이언트에 대한 연결
쓰기: 클라이언트에 응답
덱스 : 레디스 16개의 DB가 있습니다
응답 인터페이스: 응답 인터페이스
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
}5개의 응답 정의: 응답 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: 상태 응답
Int답글 : 숫자 응답
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: 데이터 길이
완료 방법: 판단 분석 완료되었나요?
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으로 구분된 일반 라인을 읽습니다. 텍스트에 rn 문자가 포함된 줄을 읽을 때 state.bulkLen과 개행 문자 rn(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에 직접: 명령을 구문 분석하고 구문 분석이 완료된 후 채널을 통해 보냅니다.
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는 사용자의 입력을 수신하고 그대로 돌려보내세요. 이제 EchoHandler를 대체하기 위해 RespHandler를 작성하고 파서가 구문 분석을 수행하도록 해야 합니다. RespHandler에는 클라이언트 연결을 관리하는 연결 구조가 있어야 합니다.
Connection: 프로토콜 레이어 핸들러에 사용되는 클라이언트 연결
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: 2차원 바이트 배열에 대한 명령 별칭
DataEntity: 문자열, 목록을 포함한 Redis 데이터를 나타냅니다. 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)
} Change 방금 작성한 내용의 주요 내용: handler.MakeHandler()
위 내용은 Golang을 기반으로 Redis 프로토콜 파서를 구현하는 방법의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!
성명：
이 기사는 yisu.com에서 복제됩니다. 침해가 있는 경우 admin@php.cn으로 문의하시기 바랍니다. 삭제