Golang 实现 Redis系列(六)如何实现 pipeline 模式的 redis 客户端
作者:Finley 发布时间:2024-05-13 10:44:23
本文的完整代码在github.com/hdt3213/godis/redis/client
通常 TCP 客户端的通信模式都是阻塞式的: 客户端发送请求 -> 等待服务端响应 -> 发送下一个请求。因为需要等待网络传输数据,完成一次请求循环需要等待较多时间。
我们能否不等待服务端响应直接发送下一条请求呢?答案是肯定的。
TCP 作为全双工协议可以同时进行上行和下行通信,不必担心客户端和服务端同时发包会导致冲突。
p.s. 打电话的时候两个人同时讲话就会冲突听不清,只能轮流讲。这种通信方式称为半双工。广播只能由电台发送到收音机不能反向传输,这种方式称为单工。
我们为每一个 tcp 连接分配了一个 goroutine 可以保证先收到的请求先先回复。另一个方面,tcp 协议会保证数据流的有序性,同一个 tcp 连接上先发送的请求服务端先接收,先回复的响应客户端先收到。因此我们不必担心混淆响应所对应的请求。
这种在服务端未响应时客户端继续向服务端发送请求的模式称为 Pipeline 模式。因为减少等待网络传输的时间,Pipeline 模式可以极大的提高吞吐量,减少所需使用的 tcp 链接数。
pipeline 模式的 redis 客户端需要有两个后台协程程负责 tcp 通信,调用方通过 channel 向后台协程发送指令,并阻塞等待直到收到响应,这是一个典型的异步编程模式。
我们先来定义 client 的结构:
type Client struct {
conn net.Conn // 与服务端的 tcp 连接
pendingReqs chan *Request // 等待发送的请求
waitingReqs chan *Request // 等待服务器响应的请求
ticker *time.Ticker // 用于触发心跳包的计时器
addr string
ctx context.Context
cancelFunc context.CancelFunc
writing *sync.WaitGroup // 有请求正在处理不能立即停止,用于实现 graceful shutdown
}
type Request struct {
id uint64 // 请求id
args [][]byte // 上行参数
reply redis.Reply // 收到的返回值
heartbeat bool // 标记是否是心跳请求
waiting *wait.Wait // 调用协程发送请求后通过 waitgroup 等待请求异步处理完成
err error
}
调用者将请求发送给后台协程,并通过 wait group 等待异步处理完成:
func (client *Client) Send(args [][]byte) redis.Reply {
request := &request{
args: args,
heartbeat: false,
waiting: &wait.Wait{},
}
request.waiting.Add(1)
client.working.Add(1)
defer client.working.Done()
client.pendingReqs <- request // 请求入队
timeout := request.waiting.WaitWithTimeout(maxWait) // 等待响应或者超时
if timeout {
return reply.MakeErrReply("server time out")
}
if request.err != nil {
return reply.MakeErrReply("request failed")
}
return request.reply
}
client 的核心部分是后台的读写协程。先从写协程开始:
// 写协程入口
func (client *Client) handleWrite() {
for req := range client.pendingReqs {
client.doRequest(req)
}
}
// 发送请求
func (client *Client) doRequest(req *request) {
if req == nil || len(req.args) == 0 {
return
}
// 序列化请求
re := reply.MakeMultiBulkReply(req.args)
bytes := re.ToBytes()
_, err := client.conn.Write(bytes)
i := 0
// 失败重试
for err != nil && i < 3 {
err = client.handleConnectionError(err)
if err == nil {
_, err = client.conn.Write(bytes)
}
i++
}
if err == nil {
// 发送成功等待服务器响应
client.waitingReqs <- req
} else {
req.err = err
req.waiting.Done()
}
}
读协程是我们熟悉的协议解析器模板, 不熟悉的朋友可以到解析Redis Cluster原理了解更多。
// 收到服务端的响应
func (client *Client) finishRequest(reply redis.Reply) {
defer func() {
if err := recover(); err != nil {
debug.PrintStack()
logger.Error(err)
}
}()
request := <-client.waitingReqs
if request == nil {
return
}
request.reply = reply
if request.waiting != nil {
request.waiting.Done()
}
}
// 读协程是个 RESP 协议解析器
func (client *Client) handleRead() error {
ch := parser.ParseStream(client.conn)
for payload := range ch {
if payload.Err != nil {
client.finishRequest(reply.MakeErrReply(payload.Err.Error()))
continue
}
client.finishRequest(payload.Data)
}
return nil
}
最后编写 client 的构造器和启动异步协程的代码:
func MakeClient(addr string) (*Client, error) {
conn, err := net.Dial("tcp", addr)
if err != nil {
return nil, err
}
ctx, cancel := context.WithCancel(context.Background())
return &Client{
addr: addr,
conn: conn,
sendingReqs: make(chan *Request, chanSize),
waitingReqs: make(chan *Request, chanSize),
ctx: ctx,
cancelFunc: cancel,
writing: &sync.WaitGroup{},
}, nil
}
func (client *Client) Start() {
client.ticker = time.NewTicker(10 * time.Second)
go client.handleWrite()
go func() {
err := client.handleRead()
logger.Warn(err)
}()
go client.heartbeat()
}
关闭 client 的时候记得等待请求完成:
func (client *Client) Close() {
// 先阻止新请求进入队列
close(client.sendingReqs)
// 等待处理中的请求完成
client.writing.Wait()
// 释放资源
_ = client.conn.Close() // 关闭与服务端的连接,连接关闭后读协程会退出
client.cancelFunc() // 使用 context 关闭读协程
close(client.waitingReqs) // 关闭队列
}
测试一下:
func TestClient(t *testing.T) {
client, err := MakeClient("localhost:6379")
if err != nil {
t.Error(err)
}
client.Start()
result = client.Send([][]byte{
[]byte("SET"),
[]byte("a"),
[]byte("a"),
})
if statusRet, ok := result.(*reply.StatusReply); ok {
if statusRet.Status != "OK" {
t.Error("`set` failed, result: " + statusRet.Status)
}
}
result = client.Send([][]byte{
[]byte("GET"),
[]byte("a"),
})
if bulkRet, ok := result.(*reply.BulkReply); ok {
if string(bulkRet.Arg) != "a" {
t.Error("`get` failed, result: " + string(bulkRet.Arg))
}
}
}
Keep working, we will find a way out.This is Finley, welcome to join us.
来源:https://www.cnblogs.com/Finley/p/14028402.html
![](https://www.aspxhome.com/images/zang.png)
![](https://www.aspxhome.com/images/jiucuo.png)
猜你喜欢
- 我们的机器学习任务通常会跟全局图像的问题有关(例如,“图像是否包含一只猫呢?”),所以我们最后一层的神经元应该对整个输入的全局敏感。通过逐渐
- 本文实例讲述了Python实现读写sqlite3数据库并将统计数据写入Excel的方法。分享给大家供大家参考,具体如下: src
- 前言MySQL提供了众多功能强大、方便易用的函数,使用这些函数,可以极大地提高用户对于数据库的管理效率,从而更加灵活地满足不同用户的需求。本
- 尽管现代浏览器已经支持了众多的CSS3属性,但是大部分设计师和开发人员貌似依然在关注于一些很“主流”的属性,如border-radius、b
- 一、前言确保安装scikit-imagenumpy二、Dataset一个例子:# 导入需要的包import torchimport torc
- 一、组件设计组件就是把图形、非图形的各种逻辑均抽象为一个统一的概念(组件)来实现开发的模式现在有一个场景,点击新增与编辑都弹框出来进行填写,
- 如果你电脑是Mac的,使用homebrew安装MySQL是一个非常便捷的方式,但是还是会出现一些问题;首先保证你已经安装了mysql,如果是
- 这篇文章主要介绍了python numpy数组复制使用实例解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,
- 最近在使用Python的过程中,发现网上很少提到在使用post方式时,怎么传一个数组作为参数的示例,此处根据自己的实践经验,给出相关示例:单
- 设置某个字段的值自增由于某个业务需要手动操作数据库,并设置主键ID不重复。做个记录set @rownum=0;select a.id, @r
- 最近笔者学会了用FrontPage XP做网页,心理特高兴,非常想把我在制作主页过程中的一些经验和大家交流交流、切磋切磋,我们一起来看看吧。
- 基于OpenCV2.4.8和 python 2.7实现简单的手势识别。以下为基本步骤 1.去除背景,提取手的轮廓2. RGB->YUV
- 深底色风格的页面设计很受欢迎,它可以创造出别致优雅、极富创造力的效果。深底色设计适用于许多网站类型,但并非所有。这种风格应该在恰当的条件下使
- 本文实例讲述了Python专用方法与迭代机制,分享给大家供大家参考之用。具体分析如下:众所周知,Python 设计哲学是“优雅”、“明确”、
- 做了一下mysql升级,开发机上mysql升级到5.6版本,结果连接一些低版本的mysql服务器报出如下异常:Warning: mysql_
- python的hashlib库中提供的hexdigest返回长度32的字符串。直接通过digest返回的16字节,有不可打印字符。问题来了,
- 本文实例讲述了Python OS模块。分享给大家供大家参考,具体如下:os模块在自动化测试中,经常需要查找操作文件,比如查找配置文件(从而读
- <?php function genpage(&$sql,$page_size=10) { global $pages,$su
- 1.分析 我们在用 php 制作网站时,分类是很重要的,在分类下面又再分类这第二个分类称为次分类,而现在大多
- 在vue项目中需要监听window窗口变化来时时计算图片的高度,于是就加了一个监听事件;确实监听到了,但是在离开当前页面进入其他页面改变窗口