learn - 项目中数据压缩传递（compress）

创作背景

有时候，服务端与前端或第三方服务交互时，会因为单次上行或者下行数据量过大，导致read或者write时耗时，间接的被误认为是服务端的处理速度问题。

目的

• 文件的压缩传输比较与选型。
• 数据传输时的压缩。

实施

文件的压缩传输比较与选型

针对目录内的文件（文本文件）进行打包压缩。

优先选择bz2方式。

# bz2压缩命令 
# tar -jcvf 生成压缩包名字  被打包压缩的目录
tar -jcvf abc-more.bz2  abc-more

#gz 打包压缩
tar -zcvf abc.en.gz  abc.en

#7z
7z a abc.en.7z abc.en

假设目录文件增加一倍：

源目录 624K

bz2打包压缩后 50K gz 打包压缩后 116K

适合场景：数据对象传输（OSS）；日志备份等。

数据传输时的压缩（发起请求和接受数据）

1. 常见的rpc交互协议：Protocol Buffers（PB）。
2. 内部外部通用且易于阅读和编写：JSON。

平时我们为了快速开发并且能快速定位问题（比如调用方直接把返回结果发给服务方，让服务方去定位自身问题）而采用JSON协议。

总之我们API这种服务中交换数据时，处理定义数据格式协议之外，还有一层可以进行优化：数据整体本身压缩。（可以想象成七层网络协议，每一层有自己的协议对数据包进行添加标记和解析协议）

// needGzip 判断数据式否上gzip
//
//      作用：
//      1.上行参数式否要进行gzip处理。
//      2.下行结果式否上gzip数据
func needGzip(hDer http.Header) bool {
        for k, v := range hDer {
                //数据上gzip的
                if k == "Content-Encoding" && len(v) > 0 {
                        if v[0] == "gzip" {
                                return true
                        }
                }
        }
        return false
}

使用gzip对api数据进行压缩，可以比较明显的减少网络数据传输。会在发送端和接收端增加“九牛一毛”的性能损耗。

type HttpClient struct {
	client *http.Client
	opts   *options
	logger Logger
	beaker *breaker.Breaker
}
// PostGzip 允许支持发起和接收数据用gzip协议，同Post作用一致
//
//	(header中需要增加Content-Encoding:gzip上行数据压缩/Accept-Encoding:zip下行解压缩 )
//	若对方服务不支持gzip的上行或者下行，切勿传相关header到该方法中
func (this *HttpClient) PostGzip(urlForw string, params map[string]string, files map[string][]HttpFormFile,
	body []byte, headers http.Header, queryParams ...map[string]string) (*HttpResponse, error) {

	logger := this.GetLogger()

	bodyBuf := &bytes.Buffer{}
	var contentType string
	if body != nil {
		bodyBuf.Write(body)

	} else if files != nil {
		bodyWriter := multipart.NewWriter(bodyBuf)
		// 普通参数
		for k, v := range params {
			fieldWriter, err := bodyWriter.CreateFormField(k)
			if err != nil {
				logger.Errorf("Post.fieldWriter.CreateFormField.k=%s.err=%s", k, err)
				return nil, err
			}

			_, err = io.Copy(fieldWriter, strings.NewReader(v))
			if err != nil {
				logger.Errorf("Post.fieldWriter.io Copy.err=%s", err)
				return nil, err
			}
		}

		// 文件参数
		for k, v := range files {
			for _, f := range v {
				fileWriter, err := bodyWriter.CreateFormFile(k, f.FileName)
				if err != nil {
					logger.Errorf("Post.fileWriter.CreateFormFile.k=%s.fileName=%s.err=%s", k, f.FileName, err)
					return nil, err
				}

				_, err = io.Copy(fileWriter, bytes.NewReader(f.FileCont))
				if err != nil {
					logger.Errorf("Post.fieldWriter.io Copy.err=%s", err)
					return nil, err
				}
			}
		}
		contentType = bodyWriter.FormDataContentType()
		bodyWriter.Close()
	} else if params != nil {
		uVals := url.Values{}
		for k, v := range params {
			uVals.Add(k, v)
		}
		bodyBuf.Write([]byte(uVals.Encode()))
		contentType = "application/x-www-form-urlencoded"
	}

	if len(queryParams) > 0 {
		urlForw = genURL(urlForw, queryParams[0])
	}

	if this.IsDebug() {
		logger.Info("request URL: ", urlForw)
	}
	var err error
	var request *http.Request
	/****gzip-begin*****/
	if needGzip(headers) {
		//数据需要gzip压缩
		gzipBody := &bytes.Buffer{}
		zw, _ := gzip.NewWriterLevel(gzipBody, gzip.BestSpeed)
		zw.Write(bodyBuf.Bytes())
		zw.Close()
		request, err = http.NewRequest("POST", urlForw, gzipBody)
	} else {
		request, err = http.NewRequest("POST", urlForw, bodyBuf)
	}

	/****gzip-end*****/
	if err != nil {
		logger.Errorf("Error Occured. %+v", err)
		return nil, err
	}

	// !!!!!
	request.Close = true

	if headers != nil {
		for k, v := range headers {
			request.Header.Set(k, v[0])
		}
	}
	if contentType != "" {
		request.Header.Set("Content-Type", contentType)
	}

	result := new(HttpResponse)
	reqDoFn := func() error {
		var response *http.Response
		response, err = this.client.Do(request)
		if err != nil || response == nil {
			logger.Errorf("PostGzip Error sending request to API. %+v", err)
			return err
		}
		defer response.Body.Close()
		result.StatusCode = response.StatusCode
		result.Headers = response.Header

		if result.StatusCode != http.StatusOK {
			//创造一个error信息
			errorText := fmt.Sprintf("this request httpCode is %d.", result.StatusCode)
			logger.Errorf(errorText)
			err = errors.New(errorText)
			return err
		}

		var dataBytes []byte
		/****响应体-解压*****/
		if needGzip(response.Header) {
			var repBuf *gzip.Reader
			repBuf, err = gzip.NewReader(response.Body)
			if err != nil {
				logger.Errorf("PostGzip Couldn't parse response body 1. %+v", err)
				return err
			}
			defer repBuf.Close()
			dataBytes, err = ioutil.ReadAll(repBuf)
		} else {
			dataBytes, err = ioutil.ReadAll(response.Body)
		}
		/****gzip-end****/
		if err != nil {
			logger.Errorf("PostGzip Couldn't parse response body 2. %+v", err)
			return err
		}

		result.Body = dataBytes
		return nil
	}

	if this.IsEnableCircuitBreaker() {
		err = this.beaker.Execute(reqDoFn)
		return result, err
	}

	err = reqDoFn()
	return result, err
}

//new
//NewClientDeadline 带有结束deadline的http
func NewClientDeadline(opts ...Option) *HttpClient {

	dopts := NewOptions(opts...)
	//deadLineTimeOut := 2 * time.Second //todo，截止时间属性 该方案不可行，因为 单例模式

	tr := &http.Transport{
		Proxy: http.ProxyFromEnvironment,
		DialContext: (&net.Dialer{
			Timeout:   defaultDialTimeout,
			KeepAlive: defaultDialKeepAlive,
			//Deadline:  time.Now().Add(deadLineTimeOut), //一个绝对超时 时间点， 作用再dail上, 错误信息： dial tcp: i/o timeout
		}).DialContext,

		MaxIdleConns:          MaxIdleConns,
		MaxIdleConnsPerHost:   MaxIdleConnsPerHost,
		IdleConnTimeout:       IdleConnTimeout,
		ResponseHeaderTimeout: dopts.timeout, //传入request后，等待接收响应时的 时间，可以认为对方 业务 程序运行时间,错误信息： net/http: timeout awaiting response headers,
	}

	return &HttpClient{
		client: &http.Client{
			CheckRedirect: nil,
			Jar:           nil,
			Transport:     tr,
			Timeout:       dopts.timeout, //context deadline exceeded (Client.Timeout exceeded while awaiting headers)
		},
		opts:   dopts,
		beaker: dopts.breaker,
	}
}

即使一次下行数据超级多（例如分页数大的情况下，或者组合的数据多的情况），也可以缩小至少5倍（500k -> 100k）。

数据传输时的压缩（接受数据请求、应答数据）

服务器可以接收gzip压缩的请求体，并将响应也以gzip格式返回。

对于接收到的请求，检查请求头中的Content-Encoding是否为gzip，如果是，则解压请求体。

检查客户端是否接受gzip响应（Accept-Encoding头部），如果接受，则将响应用gzip压缩并设置相应的响应头。

// gzipResponseWriter 包装 http.ResponseWriter 并实现 gzip 压缩
type gzipResponseWriter struct {
	io.Writer
	http.ResponseWriter
}

func (w gzipResponseWriter) Write(b []byte) (int, error) {
	return w.Writer.Write(b)
}
// gzipHandler 中间件用于处理请求和响应的 gzip 压缩
func gzipHandler(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		// 处理 gzip 压缩的请求体
		if strings.Contains(r.Header.Get("Content-Encoding"), "gzip") {
			gz, err := gzip.NewReader(r.Body)
			if err != nil {
				http.Error(w, err.Error(), http.StatusInternalServerError)
				return
			}
			defer gz.Close()
			r.Body = gz
		}

		// 检查客户端是否支持 gzip 响应
		if !strings.Contains(r.Header.Get("Accept-Encoding"), "gzip") {
			next.ServeHTTP(w, r)
			return
		}

		// 设置 gzip 响应头
		w.Header().Set("Content-Encoding", "gzip")
		gz := gzip.NewWriter(w)
		defer gz.Close()
		gzw := gzipResponseWriter{Writer: gz, ResponseWriter: w}
		next.ServeHTTP(gzw, r)
	})
}

借助middleware功能

这里采用go-zero框架的一个案例（接口Response时）：

package middleware

import (
	"compress/gzip"
	"net/http"
	"strings"
)

type gzipMiddleware struct{}

func NewGzipMiddleware() *gzipMiddleware {
	return &gzipMiddleware{}
}

func (m *gzipMiddleware) Handle(next http.HandlerFunc) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		if strings.Contains(r.Header.Get("Accept-Encoding"), "gzip") {
			w.Header().Set("Content-Encoding", "gzip")

			gw, _ := gzip.NewWriterLevel(w, gzip.BestCompression)
			defer gw.Close()
			gzr := &gzipResponseWriter{
				ResponseWriter: w,
				gzWriter:       gw,
			}
			next(gzr, r)
		} else {
			next(w, r)
		}

	}
}

type gzipResponseWriter struct {
	http.ResponseWriter
	gzWriter *gzip.Writer
}

func (gzw *gzipResponseWriter) Write(p []byte) (int, error) {
	return gzw.gzWriter.Write(p)
}

压缩前后对比：

gzip 压缩对比 （byte）,

24188 => 17136 ：压缩率29%

23464 => 16935 ： 压缩率：28%

总结

1. 如果我们有些静态文件要给使用方，可以用bz2的方式。
2. 如果纯粹服务器与服务器之间数据传输，可以使用rpc或者gzip。