golang 源碼分析: redcon 基於 redis 協議的框架
https://github.com/tidwall/redcon 是一個 Go 實現 的 Redis 兼容服務器框架。它實現了 redis 協議，封裝了網絡連接，我們可以基於這個庫快速實現一個基於 redis 協議的服務器。簡單的 redis 服務器 https://github.com/redis-go/redis 就是基於這個包實現的。
package main
import (
"log"
"strings"
"sync"
"github.com/tidwall/redcon"
)
var addr = ":6380"
func main() {
var mu sync.RWMutex
var items = make(map[string][]byte)
var ps redcon.PubSub
go log.Printf("started server at %s", addr)
err := redcon.ListenAndServe(addr,
func(conn redcon.Conn, cmd redcon.Command) {
switch strings.ToLower(string(cmd.Args[0])) {
default:
conn.WriteError("ERR unknown command '" + string(cmd.Args[0]) + "'")
case "ping":
conn.WriteString("PONG")
case "quit":
conn.WriteString("OK")
conn.Close()
case "set":
if len(cmd.Args) != 3 {
conn.WriteError("ERR wrong number of arguments for '" + string(cmd.Args[0]) + "' command")
return
}
mu.Lock()
items[string(cmd.Args[1])] = cmd.Args[2]
mu.Unlock()
conn.WriteString("OK")
case "get":
if len(cmd.Args) != 2 {
conn.WriteError("ERR wrong number of arguments for '" + string(cmd.Args[0]) + "' command")
return
}
mu.RLock()
val, ok := items[string(cmd.Args[1])]
mu.RUnlock()
if !ok {
conn.WriteNull()
} else {
conn.WriteBulk(val)
}
case "del":
if len(cmd.Args) != 2 {
conn.WriteError("ERR wrong number of arguments for '" + string(cmd.Args[0]) + "' command")
return
}
mu.Lock()
_, ok := items[string(cmd.Args[1])]
delete(items, string(cmd.Args[1]))
mu.Unlock()
if !ok {
conn.WriteInt(0)
} else {
conn.WriteInt(1)
}
case "publish":
if len(cmd.Args) != 3 {
conn.WriteError("ERR wrong number of arguments for '" + string(cmd.Args[0]) + "' command")
return
}
conn.WriteInt(ps.Publish(string(cmd.Args[1]), string(cmd.Args[2])))
case "subscribe", "psubscribe":
if len(cmd.Args) < 2 {
conn.WriteError("ERR wrong number of arguments for '" + string(cmd.Args[0]) + "' command")
return
}
command := strings.ToLower(string(cmd.Args[0]))
for i := 1; i < len(cmd.Args); i++ {
if command == "psubscribe" {
ps.Psubscribe(conn, string(cmd.Args[i]))
} else {
ps.Subscribe(conn, string(cmd.Args[i]))
}
}
}
},
func(conn redcon.Conn) bool {
// Use this function to accept or deny the connection.
// log.Printf("accept: %s", conn.RemoteAddr())
return true
},
func(conn redcon.Conn, err error) {
// This is called when the connection has been closed
// log.Printf("closed: %s, err: %v", conn.RemoteAddr(), err)
},
)
if err != nil {
log.Fatal(err)
}
}
下面看下源碼實現，源碼很簡單，主要是兩個文件：redcon/redcon.go，redcon/resp.go 前者實現了網絡連接的包裝，後者實現了 redis 協議。依賴了兩個網絡包 https://github.com/tidwall/btree，https://github.com/tidwall/match
我們還是從例子的入口函數 ListenAndServe 開始學習
func ListenAndServe(addr string,
handler func(conn Conn, cmd Command),
accept func(conn Conn) bool,
closed func(conn Conn, err error),
) error {
return ListenAndServeNetwork("tcp", addr, handler, accept, closed)
}傳入了 4 個參數，地址、服務 handler（服務核心邏輯實現的地方，處理請求並返回結果）、accept 函數和 close 函數。核心邏輯只是對 ListenAndServeNetwork 的一個包裝，確定了網絡協議是 tcp 協議
func ListenAndServeNetwork(
net, laddr string,
handler func(conn Conn, cmd Command),
accept func(conn Conn) bool,
closed func(conn Conn, err error),
) error {
return NewServerNetwork(net, laddr, handler, accept, closed).ListenAndServe()
}NewServerNetwort 函數初始化了 server，最終調用的是 server 的 ListenAndServe() 函數。
func NewServerNetwork(
net, laddr string,
handler func(conn Conn, cmd Command),
accept func(conn Conn) bool,
closed func(conn Conn, err error),
) *Server {
if handler == nil {
panic("handler is nil")
}
s := &Server{
net: net,
laddr: laddr,
handler: handler,
accept: accept,
closed: closed,
conns: make(map[*conn]bool),
}
return s
}func (s *Server) ListenAndServe() error {
return s.ListenServeAndSignal(nil)
}func (s *Server) ListenServeAndSignal(signal chan error) error {
ln, err := net.Listen(s.net, s.laddr)
if err != nil {
if signal != nil {
signal <- err
}
return err
}
s.ln = ln
if signal != nil {
signal <- nil
}
return serve(s)
}在這裏初始化了網絡連接，偵聽網絡端口，最後調用 serve 服務
func serve(s *Server) error {
for {
lnconn, err := s.ln.Accept()
if s.accept != nil && !s.accept(c) {
go handle(s, c)
}
}
}serve 是整個服務的大循環，裏面不斷 accept 請求，對每個連接，啓用一個協程去處理請求內容。
func handle(s *Server, c *conn) {
for {
cmds, err := c.rd.readCommands(nil)
for len(c.cmds) > 0 {
cmd := c.cmds[0]
s.handler(c, cmd)
}
}在 handle 函數內部調用 server 的 handler 去處理服務端請求的內容。至此整個服務端的框架基本介紹完畢。裏面還封裝了一套 TLS 的 server 邏輯，內容基本相似。
func NewServerTLS(addr string,
handler func(conn Conn, cmd Command),
accept func(conn Conn) bool,
closed func(conn Conn, err error),
config *tls.Config,
) *TLSServer {
return NewServerNetworkTLS("tcp", addr, handler, accept, closed, config)
}下面重點介紹下 handler 函數，它是 server 結構體的一個屬性
type Server struct {
mu sync.Mutex
net string
laddr string
handler func(conn Conn, cmd Command)
accept func(conn Conn) bool
closed func(conn Conn, err error)
conns map[*conn]bool
ln net.Listener
done bool
idleClose time.Duration
// AcceptError is an optional function used to handle Accept errors.
AcceptError func(err error)
}有兩個參數 Conn 網絡連接、Command 請求參數
type Conn interface {}type conn struct {
conn net.Conn
wr *Writer
rd *Reader
addr string
ctx interface{}
detached bool
closed bool
cmds []Command
idleClose time.Duration
}包裹了網絡連接和 reader、writer
redis 協議 resp 的定義如下
type RESP struct {
Type Type
Raw []byte
Data []byte
Count int
}並且也實現了相關協議的解析函數
func ReadNextRESP(b []byte) (n int, resp RESP)
func ReadNextCommand(packet []byte, argsbuf [][]byte)
func readTelnetCommand(packet []byte, argsbuf [][]byte)
func AppendAny(b []byte, v interface{}) []byteredcon 只是一個 server 框架，基於這個框架，我們可以向開發 httpserver 一樣非常方便地開發出一個兼容 redis 協議的服務端。
本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/i2JQSwo027Q4cHcXH1qL6Q