go的fasthttp学习～stackless的writer

2024-03-18 05:36
文章标签 go 学习 writer stackless fasthttp

本文主要是介绍go的fasthttp学习～stackless的writer，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

代码

package stacklessimport ("errors""fmt""io""sync""github.com/valyala/bytebufferpool"
)// Writer is an interface stackless writer must conform to.
// Writer 是 Stackless writer 必须遵守的接口
// The interface contains common subset for Writers from compress/* packages.
type Writer interface {Write(p []byte) (int, error)Flush() errorClose() errorReset(w io.Writer)
}// NewWriterFunc must return new writer that will be wrapped into
// stackless writer.
type NewWriterFunc func(w io.Writer) Writer// NewWriter creates a stackless writer around a writer returned
// from newWriter.
//
// The returned writer writes data to dstW.
// 返回的Writer将数据写入到dstW
// 使用大量栈空间的Writers被封装为stackless writer，这样可以支持更高的并发
// Writers that use a lot of stack space may be wrapped into stackless writer,
// thus saving stack space for high number of concurrently running goroutines.
//
// newWriter的意思是把内部结构体writer的私有变量 xw xWriter 通过外部的方法转换为一个stackless的成员
// 然后让内部结构体writer的zw指向它
// 后续都是通过w.zw来操作写入
func NewWriter(dstW io.Writer, newWriter NewWriterFunc) Writer {w := &writer{dstW: dstW,}w.zw = newWriter(&w.xw)return w
}type writer struct {dstW io.Writerzw   Writerxw   xWritererr errorn   intp  []byteop op
}type op intconst (opWrite op = iotaopFlushopCloseopReset
)func (w *writer) Write(p []byte) (int, error) {w.p = perr := w.do(opWrite)w.p = nilreturn w.n, err
}func (w *writer) Flush() error {return w.do(opFlush)
}func (w *writer) Close() error {return w.do(opClose)
}func (w *writer) Reset(dstW io.Writer) {// 归还writer中的bytebufferpool.ByteBufferw.xw.Reset()// 初始化writer中的zw指向的stackless的内容w.do(opReset) //nolint:errcheckw.dstW = dstW
}func (w *writer) do(op op) error {w.op = opif !stacklessWriterFunc(w) {return errHighLoad}err := w.errif err != nil {return err}// 如果 w.xw 的 bb 中还有数据，则将其写入dstWif w.xw.bb != nil && len(w.xw.bb.B) > 0 {_, err = w.dstW.Write(w.xw.bb.B)}// 更新xww.xw.Reset()return err
}var errHighLoad = errors.New("cannot compress data due to high load")var (stacklessWriterFuncOnce sync.OncestacklessWriterFuncFunc func(ctx any) bool
)func stacklessWriterFunc(ctx any) bool {stacklessWriterFuncOnce.Do(func() {stacklessWriterFuncFunc = NewFunc(writerFunc)})// 第一次执行stacklessWriterFuncFunc时// 此时macProc个子协程已经就绪，等待writerFunc执行代码return stacklessWriterFuncFunc(ctx)
}// 多协程等待执行，err存储到 w 中，外部会读取
func writerFunc(ctx any) {w := ctx.(*writer)switch w.op {case opWrite:w.n, w.err = w.zw.Write(w.p)case opFlush:w.err = w.zw.Flush()case opClose:w.err = w.zw.Close()case opReset:w.zw.Reset(&w.xw) // zw内部reset的时候，需要将w.xw的位置也重新赋值上w.err = nildefault:panic(fmt.Sprintf("BUG: unexpected op: %d", w.op))}
}type xWriter struct {bb *bytebufferpool.ByteBuffer
}func (w *xWriter) Write(p []byte) (int, error) {if w.bb == nil {w.bb = bufferPool.Get()}return w.bb.Write(p)
}func (w *xWriter) Reset() {if w.bb != nil {bufferPool.Put(w.bb)w.bb = nil}
}var bufferPool bytebufferpool.Pool

单测

package stacklessimport ("bytes""compress/flate""compress/gzip""fmt""io""testing""time"
)func TestCompressFlateSerial(t *testing.T) {t.Parallel()if err := testCompressFlate(); err != nil {t.Fatalf("unexpected error: %v", err)}
}func TestCompressFlateConcurrent(t *testing.T) {t.Parallel()if err := testConcurrent(testCompressFlate, 10); err != nil {t.Fatalf("unexpected error: %v", err)}
}func testCompressFlate() error {// 第一个参数是一个函数，将入参 w io.Writer 转换为 flate.Writer// 第二个参数是一个函数，将入参 r io.Reader 转换为 flate.Readerreturn testWriter(func(w io.Writer) Writer {// 这里是将 stackless的 xwriter 转换为 flate.Writerzw, err := flate.NewWriter(w, flate.DefaultCompression)if err != nil {panic(fmt.Sprintf("BUG: unexpected error: %v", err))}return zw}, func(r io.Reader) io.Reader {return flate.NewReader(r)})
}func TestCompressGzipSerial(t *testing.T) {t.Parallel()if err := testCompressGzip(); err != nil {t.Fatalf("unexpected error: %v", err)}
}func TestCompressGzipConcurrent(t *testing.T) {t.Parallel()if err := testConcurrent(testCompressGzip, 10); err != nil {t.Fatalf("unexpected error: %v", err)}
}func testCompressGzip() error {return testWriter(func(w io.Writer) Writer {return gzip.NewWriter(w)}, func(r io.Reader) io.Reader {zr, err := gzip.NewReader(r)if err != nil {panic(fmt.Sprintf("BUG: cannot create gzip reader: %v", err))}return zr})
}// 第一个参数是一个函数，将入参 w io.Writer 包裹为 flate.Writer
// 第二个参数是一个函数，将入参 r io.Reader 包裹为 flate.Reader
func testWriter(newWriter NewWriterFunc, newReader func(io.Reader) io.Reader) error {dstW := &bytes.Buffer{}// 生成一个stackless的Writer w// 并调用newWriter进行初始化：w.zw = newWriter(&w.xw) w.xw是一个xWriter// type xWriter struct {//	  bb *bytebufferpool.ByteBuffer// }//// 说白了就是将 w.xw 作为flate.Writer的底层writer来制作一个新的flate.Writer，并让 w.zw 指向这个 flate.Writer// 利用stackless的Writer去操作自己的w.zw，进而实现flate.Writer操作w.xw的目的w := NewWriter(dstW, newWriter)for i := 0; i < 5; i++ {// 这里明明已经将dstW赋值给stackless的writer了，又将其作为参数传入// 是为了通过dstW读取值，判断写入writer的内容是对的if err := testWriterReuse(w, dstW, newReader); err != nil {return fmt.Errorf("unexpected error when re-using writer on iteration %d: %w", i, err)}dstW = &bytes.Buffer{}w.Reset(dstW)}return nil
}func testWriterReuse(w Writer, r io.Reader, newReader func(io.Reader) io.Reader) error {wantW := &bytes.Buffer{}// creates a writer that duplicates its writes to all the provided writers// similar to the Unix tee(1) command// 写入多个writer，如果一个失败，则全部返回失败mw := io.MultiWriter(w, wantW)for i := 0; i < 30; i++ {fmt.Fprintf(mw, "foobar %d\n", i)// 屏蔽掉也没问题if i%13 == 0 {// 实质是对 w.zw的Flush()// if err := w.Flush(); err != nil {// 	return fmt.Errorf("error on flush: %w", err)// }}}// 实质是对 w.zw的Close()// 没有Close和Flush的话，数据不会被刷入stackless的xWriter中w.Close()zr := newReader(r)data, err := io.ReadAll(zr)if err != nil {return fmt.Errorf("unexpected error: %w, data=%q", err, data)}// 从两个writer中读出来的结果相同wantData := wantW.Bytes()if !bytes.Equal(data, wantData) {return fmt.Errorf("unexpected data: %q. Expecting %q", data, wantData)}return nil
}func testConcurrent(testFunc func() error, concurrency int) error {ch := make(chan error, concurrency)for i := 0; i < concurrency; i++ {go func() {ch <- testFunc()}()}for i := 0; i < concurrency; i++ {select {case err := <-ch:if err != nil {return fmt.Errorf("unexpected error on goroutine %d: %w", i, err)}case <-time.After(time.Second):return fmt.Errorf("timeout on goroutine %d", i)}}return nil
}// 到底什么是flush
// Flush flushes any pending data to the underlying writer.
// 将任何挂起的数据写入底层的writer (underlying writer)
// It is useful mainly in compressed network protocols, to ensure that
// a remote reader has enough data to reconstruct a packet.
// 在压缩网络协议（compressed network protocols）中很有用
// 确保一个远端的reader有足够数据去重构一个包
//
// Flush does not return until the data has been written.
// Calling Flush when there is no pending data still causes the Writer
// to emit a sync marker of at least 4 bytes.
// 发出至少4个字节的同步标记（emit a sync marker）
// If the underlying writer returns an error, Flush returns that error.
//
// In the terminology of the zlib library, Flush is equivalent to Z_SYNC_FLUSH.// flushes and closes the writer.
// close前先flushes writer
//
// func (w *Writer) Close() error {
//     return w.d.close()
// }

运用


var flateReaderPool sync.Poolfunc acquireStacklessGzipWriter(w io.Writer, level int) stackless.Writer {nLevel := normalizeCompressLevel(level)p := stacklessGzipWriterPoolMap[nLevel]v := p.Get()if v == nil {// stackless.NewWriter 第二个参数是一个函数// 用于将一个具体的 *gzip.Writer 赋值给  stackless.Writer 接口//// newWriter的意思是把内部结构体writer的私有变量 xw xWriter 通过外部的方法转换为一个stackless的成员// 然后让内部结构体writer的zw指向它// 后续都是通过w.zw来操作写入return stackless.NewWriter(w, func(w io.Writer) stackless.Writer {// acquireRealGzipWriter：从资源池取一个/new一个gzipWriter并resetreturn acquireRealGzipWriter(w, level)})}sw := v.(stackless.Writer)sw.Reset(w)return sw
}func releaseStacklessGzipWriter(sw stackless.Writer, level int) {sw.Close()nLevel := normalizeCompressLevel(level)p := stacklessGzipWriterPoolMap[nLevel]p.Put(sw)
}// acquireRealGzipWriter
// 从资源池取一个/new一个gzipWriter并reset
func acquireRealGzipWriter(w io.Writer, level int) *gzip.Writer {nLevel := normalizeCompressLevel(level)p := realGzipWriterPoolMap[nLevel]v := p.Get()if v == nil {zw, err := gzip.NewWriterLevel(w, level)if err != nil {// gzip.NewWriterLevel only errors for invalid// compression levels. Clamp it to be min or max.if level < gzip.HuffmanOnly {level = gzip.HuffmanOnly} else {level = gzip.BestCompression}zw, _ = gzip.NewWriterLevel(w, level)}return zw}zw := v.(*gzip.Writer)zw.Reset(w)return zw
}

总结

如此精妙的设计，真的太厉害了
利用stackless的NewWriter，来实现接管一个真正的复杂的高层Writer
复杂的高层Writer利用stackless的writer提供的xWriter来做它的底层io，将编码后的数据存入里面
stackless的Flush和Close其实就是接管的复杂的高层Writer的这两，
最终stackless的worker将上述结果存入 dstWriter 中

这篇关于go的fasthttp学习～stackless的writer的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/821325。

相关文章

GO语言短变量声明的实现示例

GO语言短变量声明的实现示例

《GO语言短变量声明的实现示例》在Go语言中,短变量声明是一种简洁的变量声明方式,使用:=运算符,可以自动推断变量类型,下面就来具体介绍一下如何使用,感兴趣的可以了解一下... 目录基本语法功能特点与var的区别适用场景注意事项基本语法variableName := value功能特点1、自动类型推
阅读更多...
GO语言中函数命名返回值的使用

GO语言中函数命名返回值的使用

《GO语言中函数命名返回值的使用》在Go语言中,函数可以为其返回值指定名称,这被称为命名返回值或命名返回参数,这种特性可以使代码更清晰,特别是在返回多个值时,感兴趣的可以了解一下... 目录基本语法函数命名返回特点代码示例命名特点基本语法func functionName(parameters) (nam
阅读更多...
Go之errors.New和fmt.Errorf 的区别小结

Go之errors.New和fmt.Errorf 的区别小结

《Go之errors.New和fmt.Errorf的区别小结》本文主要介绍了Go之errors.New和fmt.Errorf的区别,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考... 目录error的基本用法1. 获取错误信息2. 在条件判断中使用基本区别1.函数签名2.使用场景详细对
阅读更多...
Unity新手入门学习殿堂级知识详细讲解（图文）

Unity新手入门学习殿堂级知识详细讲解（图文）

《Unity新手入门学习殿堂级知识详细讲解（图文）》Unity是一款跨平台游戏引擎,支持2D/3D及VR/AR开发,核心功能模块包括图形、音频、物理等,通过可视化编辑器与脚本扩展实现开发,项目结构含A... 目录入门概述什么是 UnityUnity引擎基础认知编辑器核心操作Unity 编辑器项目模式分类工程
阅读更多...
Go语言连接MySQL数据库执行基本的增删改查

Go语言连接MySQL数据库执行基本的增删改查

《Go语言连接MySQL数据库执行基本的增删改查》在后端开发中,MySQL是最常用的关系型数据库之一,本文主要为大家详细介绍了如何使用Go连接MySQL数据库并执行基本的增删改查吧... 目录Go语言连接mysql数据库准备工作安装 MySQL 驱动代码实现运行结果注意事项Go语言执行基本的增删改查准备工作
阅读更多...
Go中select多路复用的实现示例

Go中select多路复用的实现示例

《Go中select多路复用的实现示例》Go的select用于多通道通信,实现多路复用,支持随机选择、超时控制及非阻塞操作,建议合理使用以避免协程泄漏和死循环,感兴趣的可以了解一下... 目录一、什么是select基本语法：二、select 使用示例示例1：监听多个通道输入三、select的特性四、使用se
阅读更多...
Go语言使用Gin处理路由参数和查询参数

Go语言使用Gin处理路由参数和查询参数

《Go语言使用Gin处理路由参数和查询参数》在WebAPI开发中,处理路由参数（PathParameter）和查询参数（QueryParameter）是非常常见的需求,下面我们就来看看Go语言... 目录一、路由参数 vs 查询参数二、Gin 获取路由参数和查询参数三、示例代码四、运行与测试1. 测试编程路
阅读更多...
Python学习笔记之getattr和hasattr用法示例详解

Python学习笔记之getattr和hasattr用法示例详解

《Python学习笔记之getattr和hasattr用法示例详解》在Python中,hasattr()、getattr()和setattr()是一组内置函数,用于对对象的属性进行操作和查询,这篇文章... 目录1.getattr用法详解1.1 基本作用1.2 示例1.3 原理2.hasattr用法详解2.
阅读更多...
Go语言使用net/http构建一个RESTful API的示例代码

Go语言使用net/http构建一个RESTful API的示例代码

《Go语言使用net/http构建一个RESTfulAPI的示例代码》Go的标准库net/http提供了构建Web服务所需的强大功能,虽然众多第三方框架（如Gin、Echo）已经封装了很多功能,但... 目录引言一、什么是 RESTful API？二、实战目标：用户信息管理 API三、代码实现1. 用户数据
阅读更多...
Go语言网络故障诊断与调试技巧

Go语言网络故障诊断与调试技巧

《Go语言网络故障诊断与调试技巧》在分布式系统和微服务架构的浪潮中,网络编程成为系统性能和可靠性的核心支柱,从高并发的API服务到实时通信应用,网络的稳定性直接影响用户体验,本文面向熟悉Go基本语法和... 目录1. 引言2. Go 语言网络编程的优势与特色2.1 简洁高效的标准库2.2 强大的并发模型2.
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2