Maison > Article > base de données > Comment implémenter l'analyseur de protocole Redis basé sur Golang
Comment implémenter l'analyseur de protocole Redis basé sur Golang
- WBOYavant
- 2023-05-28 19:13:21727parcourir
Protocole RESP
RESP est un protocole de communication entre le client et le serveur. Il existe cinq formats :
Réponse normale : une chaîne commençant par "+" et se terminant par "rn"
Réponse d'erreur : commençant par. " -" sous forme de chaîne et se termine par "rn"
Entier : commence par " : " et se termine par "rn" sous forme de chaîne
Chaîne multiligne : commence par "$" suivi des octets réels envoyés Numéro, commençant et se terminant par "rn"
$3rnabcrn
Array : commençant par "*", suivi du nombre de membres
SET valeur clé
*3rn$3rnSETrn$3 rn keyrn$5rnvaluern
Les commandes ou données envoyées par le client et le serveur doivent utiliser rn (CRLF) comme caractère de nouvelle ligne.
Lorsque nous saisissons une série de commandes comme *3rn$3rnSETrn$3rnkeyrn$5rnvaluern, le serveur reçoit la commande suivante :
*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
}Interface de connexion : une connexion pour le client Redis
Write : réponse au client
GetDBInde x : Redis a 16 bases de données
Interface de réponse : Interface de réponse
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
}Définissez cinq réponses : réponse pong, ok, null, tableau vide, vide
resp/reply/ réponse.go
type ErrorReply interface {
Error() string
ToBytes() []byte
}ErrorReply : Définir l'interface d'erreur
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 définit 5 types d'erreurs : erreur inconnue UnknownErrReply, erreur de numéro de paramètre ArgNumErrReply, erreur de syntaxe SyntaxErrReply, Erreur de type de données de réponse , Erreur de protocole 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 : réponse à une chaîne
MultiBulkReply : réponse à un tableau de chaînes
-
StatusReply : réponse d'état
IntRéponse : Réponse numérique
StandardErrReply : Réponse d'erreur standard
IsErrorReply : Détermine s'il s'agit d'une réponse d'erreur
ToBytes : Convertit la chaîne dans le format spécifié par le protocole RESP
resp/ analyseur /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) {
......
}Structure de charge utile : les données qui nous sont envoyées par le service client
Reply : les données envoyées par le client et le serveur entre eux sont appelées Reply
readState structure :
readingMultiLine : analyse d'une seule ligne ou de données multilignes
Méthode terminée : analyse du jugement Est-elle terminée ? -
Méthode ParseStream : analyse les données de manière asynchrone et les met dans le pipeline, et renvoie les données du pipeline
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 : lit ligne par ligne. Lire les lignes normales, séparées par n. Lors de la lecture d'une ligne contenant le caractère rn dans le texte, state.bulkLen plus le caractère de nouvelle ligne 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 : analysez l'en-tête du tableau, définissez le nombre de lignes attendu et les paramètres associés.
parseBulkHeader : analyse l'en-tête d'une chaîne multiligne. 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 : analyser les commandes sur une seule ligne
readBody : lire les commandes multilignes, définissez BulkLen lors de la lecture de la ligne suivante, utilisez ce +2. S'il ne commence pas par $, ajoutez-le. directement à argsfunc 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 : analysez la commande et envoyez-la via le canal une fois l'analyse terminée
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
}L'EchoHandler que nous avons écrit auparavant est utilisé pour recevoir l'entrée de l'utilisateur et remettez-le intact. Vous devez maintenant écrire un RespHandler pour remplacer le EchoHandler et laisser l'analyseur effectuer l'analyse. Une structure de connexion qui gère les connexions client doit exister dans RespHandler.
Connexion : connexion client, utilisée dans le gestionnaire de couche de protocole
resp/handler/handler.govar (
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 : similaire à l'écho précédent, avec l'ajout de la couche principale db.exec pour effectuer l'analyse de la commande
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 : exécution de la couche principale AfterClientClose : méthode de suivi après la fermeture
CmdLine : alias de commande pour un tableau d'octets bidimensionnel
DataEntity : représente les données Redis, y compris la chaîne, la liste, set, etc.
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 : tester la couche de protocole
Exec : une fois la commande analysée, utilisez MakeMultiBulkReply pour l'envelopper et la renvoyer
main.goerr := tcp.ListenAndServeWithSignal(
&tcp.Config{
Address: fmt.Sprintf("%s:%d",
config.Properties.Bind,
config.Properties.Port),
},
handler.MakeHandler())
if err != nil {
logger.Error(err)
} Change principal de ce que vous venez d'écrire : handler.MakeHandler()
Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!