分布式缓存原理及Go实现 - 第二章

第二章：基于TCP的内存缓存服务

前面的第1章实现了基于HTTP/REST的内存内缓存服务，利用了Go语言本身的HTTP框架，其性能不如Redis。本章要实现基于TCP的缓存服务来提升性能。

2.1 基于TCP的缓存协议规范

2.1.1 协议规范

此外curl命令只适用于HTTP客户端，我们需要一个能直接使用的TCP的客户端，本章自己实现了一个。

对于TCP来说，客户端和服务端之间传输的是网络字节流，它不区分方法及URL，头部和正文这些格式化的文本。

TCP的ABDF表达：SP表示空格，DIGIT表示0~9；OCTET表示八位数据，也就是一个字节，更多ABDF表达规范参考：

网络协议-HTTP 协议(DNS解析和ABNF描述)_爱因诗贤的博客-CSDN博客

2.1.2 缓存Set流程

1. 客户端发出command字节流：S<klen><SP><vlen><SP><key><value>
2. 服务端解析command并提取出 key 和 value，然后调用inMemoryCache.Set将键值对保存在内存的map中。
3. 如果cache.Cache.Set方法成功，就将tcp.Server向客户端连接写入”0”.

2.1.3 缓存Get流程

1. 客户端发出command字节流：G<klen><SP><key>
2. 服务端解析command并提取出 key，然后调用inMemoryCache.Get方法查询key对应的value。
3. 如果cache.Cache.Get方法错误，tcp.Server就向客户端写入一个error.

2.1.4 缓存Del流程

1. 客户端发出command字节流：D<klen><SP><key>
2. 服务端解析command并提取出 key，然后调用inMemoryCache.Del方法删除key对应的value。
3. 如果cache.Cache.Get方法错误，tcp.Server就向客户端写入一个error.

2.2 Go语言实现

2.2.1 main函数

package main
import (
	"./cache"
	"./http"
	"./tcp"
)

func main() {
	ca := cache.New("inmemory")
	go tcp.New(ca).Listen()	//调用goroutine多线程并发
	http.New(ca).Listen()
}

2.2.2 TCP包的实现

Server结构体的实现

package tcp

import (
	"net"

	"../cache"
)

type Server struct {
	cache.Cache
}

func (s *Server) Listen() {
	l, e := net.Listen("tcp", ":12346") //监听本机TCP的12346端口
	if e != nil {
		panic(e)
	}
	for {
		c, e := l.Accept() //无限循环接收客户端的连接并调用Server.process处理这个连接
		if e != nil {
			panic(e)
		}
		go s.process(c) //process函数定义再process.go中，Server.process运行在一个新的goroutine上，所以原来的goroutine可以继续执行
	}
}

func New(c cache.Cache) *Server {
	return &Server{c}
}

Server.process方法的相关实现

//readkey.go
package tcp

import (
	"bufio"
	"io"
)

func (s *Server) readKey(r *bufio.Reader) (string, error) { //Reader实现了给io接口对象附加缓冲
	klen, e := readLen(r) //以空格为分隔符并将其转换为一个整数
	if e != nil {
		return "", e
	}
	k := make([]byte, klen)
	_, e = io.ReadFull(r, k)
	if e != nil {
		return "", e
	}
	return string(k), nil
}

func (s *Server) readKeyAndValue(r *bufio.Reader) (string, []byte, error) {
	klen, e := readLen(r)
	if e != nil {
		return "", nil, e
	}
	vlen, e := readLen(r)
	if e != nil {
		return "", nil, e
	}
	k := make([]byte, klen)
	_, e = io.ReadFull(r, k)
	if e != nil {
		return "", nil, e
	}
	v := make([]byte, vlen)
	_, e = io.ReadFull(r, v)
	if e != nil {
		return "", nil, e
	}
	return string(k), v, nil
}

//utils.go
package tcp

import (
	"bufio"
	"fmt"
	"net"
	"strconv"
	"strings"
)

func readLen(r *bufio.Reader) (int, error) {
	tmp, e := r.ReadString(' ')
	if e != nil {
		return 0, e
	}
	l, e := strconv.Atoi(strings.TrimSpace(tmp))
	if e != nil {
		return 0, e
	}
	return l, nil
}

func sendResponse(value []byte, err error, conn net.Conn) error {
	if err != nil {
		errString := err.Error()
		tmp := fmt.Sprintf("-%d ", len(errString)) + errString
		_, e := conn.Write([]byte(tmp))
		return e
	}
	vlen := fmt.Sprintf("%d ", len(value))
	_, e := conn.Write(append([]byte(vlen), value...))
	return e
}

//process.go
package tcp

import (
	"bufio"
	"io"
	"log"
	"net"
)

func (s *Server) get(conn net.Conn, r *bufio.Reader) error {
	k, e := s.readKey(r)
	if e != nil {
		return e
	}
	v, e := s.Get(k)
	return sendResponse(v, e, conn) //sendResponse函数根据参数将服务端的error或者value写入客户端连接
}

func (s *Server) set(conn net.Conn, r *bufio.Reader) error {
	k, v, e := s.readKeyAndValue(r)
	if e != nil {
		return e
	}
	return sendResponse(nil, s.Set(k, v), conn)
}

func (s *Server) del(conn net.Conn, r *bufio.Reader) error {
	k, e := s.readKey(r)
	if e != nil {
		return e
	}
	return sendResponse(nil, s.Del(k), conn)
}

func (s *Server) process(conn net.Conn) {
	defer conn.Close()         //函数的最后关闭
	r := bufio.NewReader(conn) //对客户端连接进行缓冲读取，因为网络不稳定，在我们进行读取时，客户端的数据可能只读了一半，我们希望可以阻塞等待
	for {                      //无限循环获取command的op方法
		op, e := r.ReadByte()
		if e != nil {
			if e != io.EOF {
				log.Println("close connection due to error:", e)
			}
			return
		}
		if op == 'S' {
			e = s.set(conn, r)
		} else if op == 'G' {
			e = s.get(conn, r)
		} else if op == 'D' {
			e = s.del(conn, r)
		} else {
			log.Println("close connection due to invalid operation:", op)
			return
		}
		if e != nil {
			log.Println("close connection due to error:", e)
			return
		}
	}
}

2.2.3 客户端的实现

package main

import (
	"flag"
	"fmt"
	"go-implement-your-cache-server-master/cache-benchmark/cacheClient"
)

func main() {
	server := flag.String("h", "localhost", "cache server address") //flag的用法是首先创建所有需要解析的变量，然后调用flag.Phase函数解析命令行
	op := flag.String("c", "get", "command, could be get/set/del")  //"c"代表-c代表传入此参数,第二个参数代表默认传参，第三个参数为打印给用户的提示信息
	key := flag.String("k", "", "key")
	value := flag.String("v", "", "value")
	flag.Parse()
	client := cacheClient.New("tcp", *server)
	cmd := &cacheClient.Cmd{*op, *key, *value, nil}
	client.Run(cmd)
	if cmd.Error != nil {
		fmt.Println("error:", cmd.Error)
	} else {
		fmt.Println(cmd.Value)
	}
}

编译和运行客户端的命令：

go build
./client --help

#得到如下的结果
  -c string
        command, could be get/set/del (default "get")
  -h string
        cache server address (default "localhost")
  -k string
        key
  -v string
        value

本书后续用到的TCP客户端都在此运行。

编译并运行chapter2中的server之后就可以在客户端进行测试了。

./client -c set -k testkey -v testvalue #上传一个键值对
curl 127.0.0.1:12345/status	#查看服务端的状态

服务端的状态如下：

StatusCode        : 200
StatusDescription : OK
Content           : {"Count":1,"KeySize":7,"ValueSize":9}
RawContent        : HTTP/1.1 200 OK
                    Content-Length: 37
                    Content-Type: text/plain; charset=utf-8
                    Date: Fri, 01 Apr 2022 09:27:09 GMT

                    {"Count":1,"KeySize":7,"ValueSize":9}
Forms             : {}
Headers           : {[Content-Length, 37], [Content-Type, text/plain; charset=utf-8], [Date, Fri, 01 Apr 2022 09:27:09 GMT]}
Images            : {}
InputFields       : {}
Links             : {}
ParsedHtml        : mshtml.HTMLDocumentClass
RawContentLength  : 37

./cache-benchmark -type tcp -n 100000 -r 100000 -t set
./cache-benchmark -type tcp -n 100000 -r 100000 -t get

检测性能发现使用TCP的缓存确实可以提升一定的性能。