前言

　　最近，一个数据查询服务被业务方反馈拿不到数据，但接口响应是成功的，不报错，仔细排查后发现数据查询库用的是 pgx，但 pgx 返回的错误未被处理，导致服务接口没有响应错误。

　　在后续的排查过程中，发现这其实不算是 pgx 的问题，而是 database/sql 中的坑，所有涉及用 database/sql 查询的都需要显式处理 rows.Err()。

问题篇

　　服务所用 pgx 版本为 4.10.1。查询函数主要用的是 QueryRow（返回一条数据） 和 Query（返回多条数据），更近一步的测试中（人为制造查询错误，eg：锁表）发现，调用 QueryRow 函数的接口，如果发生查询错误的问题，服务接口会正常响应错误。深入 pgx 源码发现，QueryRow 本质是对 Query 的进一步封装，对应的 Scan 函数源码为：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104

// QueryRow 返回的 connRow 对应继承自 Query 返回的 connRows 对象
type connRow connRows
func (r *connRow) Scan(dest ...interface{}) (err error) {
    // 显式类型转换
    rows := (*connRows)(r)

    if rows.Err() != nil {
        return rows.Err()
    }

    if !rows.Next() {
        if rows.Err() == nil {
            return ErrNoRows
        }
        return rows.Err()
    }

    rows.Scan(dest...)
    rows.Close()
    // 关键的错误
    return rows.Err()
}

func (rows *connRows) Scan(dest ...interface{}) error {
    ci := rows.connInfo
    fieldDescriptions := rows.FieldDescriptions()
    values := rows.values

    if len(fieldDescriptions) != len(values) {
        err := errors.Errorf("number of field descriptions must equal number of values, got %d and %d", len(fieldDescriptions), len(values))
        rows.fatal(err)
        return err
    }
    if len(fieldDescriptions) != len(dest) {
        err := errors.Errorf("number of field descriptions must equal number of destinations, got %d and %d", len(fieldDescriptions), len(dest))
        rows.fatal(err)
        return err
    }

    if rows.scanPlans == nil {
        rows.scanPlans = make([]pgtype.ScanPlan, len(values))
        for i := range dest {
            rows.scanPlans[i] = ci.PlanScan(fieldDescriptions[i].DataTypeOID, fieldDescriptions[i].Format, dest[i])
        }
    }

    for i, dst := range dest {
        if dst == nil {
            continue
        }

        err := rows.scanPlans[i].Scan(ci, fieldDescriptions[i].DataTypeOID, fieldDescriptions[i].Format, values[i], dst)
        if err != nil {
            err = scanArgError{col: i, err: err}
            rows.fatal(err)
            return err
        }
    }

    // 由于返回多行数据，需要多次 Scan， 所以不能返回 rows.Err()
    return nil
}

// 有些数据库查询错误，只有在 Close 之后，再调用 rows.Err() 捕获
func (rows *connRows) Close() {
    if rows.closed {
        return
    }

    rows.closed = true

    if rows.resultReader != nil {
        var closeErr error
        rows.commandTag, closeErr = rows.resultReader.Close()
        if rows.err == nil {
            // 赋值错误
            rows.err = closeErr
        }
    }

    if rows.multiResultReader != nil {
        closeErr := rows.multiResultReader.Close()
        if rows.err == nil {
            // 赋值错误
            rows.err = closeErr
        }
    }

    if rows.logger != nil {
        if rows.err == nil {
            if rows.logger.shouldLog(LogLevelInfo) {
                endTime := time.Now()
                rows.logger.log(rows.ctx, LogLevelInfo, "Query", map[string]interface{}{"sql": rows.sql, "args": logQueryArgs(rows.args), "time": endTime.Sub(rows.startTime), "rowCount": rows.rowCount})
            }
        } else {
            if rows.logger.shouldLog(LogLevelError) {
                rows.logger.log(rows.ctx, LogLevelError, "Query", map[string]interface{}{"err": rows.err, "sql": rows.sql, "args": logQueryArgs(rows.args)})
            }
            if rows.err != nil && rows.conn.stmtcache != nil {
                rows.conn.stmtcache.StatementErrored(rows.sql, rows.err)
            }
        }
    }
}

　　从源码中可看出 QueryRow 的 Scan 函数有一系列的错误处理，而 Query 对应的 Scan 是更底层的函数，返回的仅是 Scan 过程中的错误，其他的错误需要在业务上层处理。Close 函数同样可能会出现错误，需要调用 rows.Err() 主动检查错误（这一步至关重要）。对于 Close 报的错，可以这样处理：

1
2
3
4
5
6
7
8
9
10
11
12

var err error
defer func() {
    rows.Close()
    closeErr := rows.Err()
    if err != nil {
        if closeErr != nil {
            log.Printf("failed to close rows: %v", err)
        }
        return
    }
    err = closeErr
}

　　数据库执行 sql 失败的错误（eg：canceling statement due to conflict with recovery），在 Close 后才会暴露出来，所以不处理这个错误，就不会返回错误，但数据又查不到，服务接口也表现为响应成功，导致上层业务误认为数据库里还真没数据。

　　最好的方式还是避免每次都手动 Scan，pgx 其实还提供了更上层的函数 QueryFunc，该函数封装了大部分错误处理：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

func (c *Conn) QueryFunc(ctx context.Context, sql string, args []interface{}, scans []interface{}, f func(QueryFuncRow) error) (pgconn.CommandTag, error) {
    rows, err := c.Query(ctx, sql, args...)
    if err != nil {
        return nil, err
    }
    defer rows.Close()
  
    // 最后一次 Next() 会自动调用 Close()
    for rows.Next() {
        err = rows.Scan(scans...)
        if err != nil {
            return nil, err
        }

        err = f(rows)
        if err != nil {
            return nil, err
        }
    }

    if err := rows.Err(); err != nil {
        return nil, err
    }

    return rows.CommandTag(), nil
}

不过 QueryFunc 函数在新版本中（5.7.2）已被 ForEachRow 替代：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

func ForEachRow(rows Rows, scans []any, fn func() error) (pgconn.CommandTag, error) {
    defer rows.Close()

    for rows.Next() {
        err := rows.Scan(scans...)
        if err != nil {
            return pgconn.CommandTag{}, err
        }

        err = fn()
        if err != nil {
            return pgconn.CommandTag{}, err
        }
    }

    if err := rows.Err(); err != nil {
        return pgconn.CommandTag{}, err
    }

    return rows.CommandTag(), nil
}

而 ForEachRow 的使用示例可以看这个函数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

func (c *Conn) getCompositeFields(ctx context.Context, oid uint32) ([]pgtype.CompositeCodecField, error) {
    var typrelid uint32

    err := c.QueryRow(ctx, "select typrelid from pg_type where oid=$1", oid).Scan(&typrelid)
    if err != nil {
        return nil, err
    }

    var fields []pgtype.CompositeCodecField
    var fieldName string
    var fieldOID uint32
    rows, _ := c.Query(ctx, `select attname, atttypid
from pg_attribute
where attrelid=$1
    and not attisdropped
    and attnum > 0
order by attnum`,
        typrelid,
    )
    // 这里是示例
    _, err = ForEachRow(rows, []any{&fieldName, &fieldOID}, func() error {
        dt, ok := c.TypeMap().TypeForOID(fieldOID)
        if !ok {
            return fmt.Errorf("unknown composite type field OID: %v", fieldOID)
        }
        fields = append(fields, pgtype.CompositeCodecField{Name: fieldName, Type: dt})
        return nil
    })
    if err != nil {
        return nil, err
    }

    return fields, nil
}

后记

　　对于一个不熟悉的底层库，最好的学习方式还是看它的示例代码，库的开发者很难知道用户会踩哪些坑，文档中自然不会有，毕竟当局者迷。只从文档出发，很容易陷进未知的坑里，甚至掉坑里都不知道，业务出问题后，花费大代价排查之后，才知道掉坑里了。陌生的开源库在使用的时候还是先全库 clone 下来，用 api 的时候，就去源码里搜一下，看看开发者写的示例（不管是测试，还是其他地方的调用），当然现在也可以让 AI 先写，人只要再核实一下文档和源码，能节省很多学习的功夫。

前言

　　最近看到这么一个工具——ollama-ocr，利用本地大模型直接进行 OCR，试用了一下，感觉效果还不错，联想之前看到的一个项目——PDFMathTranslate，感觉本地大模型已经非常成熟了，部署使用也越来越简单了。

如果需要提供 LLM 服务，还是用 vLLM 部署相对合适。

配置篇

　　硬件：M3 MBP，10 核 GPU（Metal 3），16G 内存

　　系统&软件：macOS Sonoma 14.5，Ollama 0.5.4，AnythingLLM 1.7.2；

体验篇

ollama 篇

　　第一次启动 Ollama 时，会出现安装「command line」的引导界面，输入电脑用户密码安装即可，安装命令行之后，可以直接点击「Finish」，之后通过终端命令 ollama 操作 Ollama，毕竟 Ollama 没有提供 UI 界面，当然也有很多三方的界面。

　　之后在终端中直接运行 ollama run llama3.2:3b，等待模型拉取完成后，即可直接与模型对话。之后可以在 Ollama 官网 Models 搜索尝试更多模型。

命令解析：

• ollama pull [model:tag]：拉取模型；
• ollama run [model:tag]：运行拉取的模型，若没有，会自动拉取之后运行；
• ollama list：查看全部已拉取的模型；
• ollama show [model:tag]：显示模型信息；
• ollama rm [model:tag]：删除已拉取的模型；
• ollama ps：查看正在运行的模型；
• ollama stop [model:tag]：停止正在运行的模型；

　　Mac 中修改全局环境变量可通过 launchctl setenv 命令，eg： launchctl setenv OLLAMA_ORIGINS "*"，允许 ollama 请求跨域。

　　Mac 中 ollama 拉取的模型文件默认放在 ~/.ollama/models 目录中，可通过修改 OLLAMA_MODELS 环境变量更改模型安装目录。

　　Ollama 默认服务地址端口是：127.0.0.1:11434，Mac 中查看进程监听的端口号命令为 lsof -nP -p <pid>。可通过修改 OLLAMA_HOST 环境变量更改默认端口，eg：launchctl setenv OLLAMA_HOST "0.0.0.0:6006"。

　　默认情况下，运行模型后，如果 5 分钟未与模型进行交互，将会自动停止该模型。

　　Mac 启动 Ollama 后，会在菜单栏上出现一个羊驼图标，但有时这个图标会被“刘海”挡住，导致无法退出 Ollama，这时可以使用 osascript -e 'tell app "Ollama" to quit' 命令退出 Ollama。

　　对于开发者，若需要更改默认端口，需修改环境变量：export OLLAMA_HOST=0.0.0.0:6006（如此可将 ollama 的默认端口修改为 6006），之后通过 ollama serve 命令启动 Ollama，通过该命令启动的不会在菜单栏上出现羊驼图标。若出现跨域问题，同样需要修改环境变量：export OLLAMA_ORIGINS="*"。

AnythingLLM 篇

　　AnythingLLM 有两种安装模式，一种是桌面版，一种是 Docker 版，桌面版只能本地使用，Docker 版相当于是服务版，支持多用户云端使用。本次选用的桌面版，基本的 RAG 功能也都有。Docker vs Desktop Version

　　第一次启动 AnythingLLM 时，有一些设置引导，设置「LLM 偏好」时选择 Ollama，其他的都默认即可。创建工作空间之后就可以上传本地文件，建立自己的知识库。

　　Shaun 在使用中感觉，AnythingLLM 响应还是比较慢，分析/提炼/归纳/总结本地文档的速度有限。可能是机器配置还是有点低了。

PDFMathTranslate 篇

　　命令行使用：

1
2
3
4
5

# 用 ollama 将本地文件 Black Hat Rust.pdf 从英文翻译为中文
pdf2zh Black\ Hat\ Rust.pdf -s ollama -li en -lo zh

# 启动 pdf2zh 网页，免去命令行使用 Web 页面设置翻译参数
pdf2zh -i

　　出现 NotOpenSSLWarning: urllib3 v2 only supports OpenSSL 1.1.1+, currently the 'ssl' module is compiled with 'LibreSSL 2.8.3' 警告时可忽略，不影响使用。

　　如果出现 huggingface_hub.errors.LocalEntryNotFoundError 错误，需要配置 hugging_face 国内镜像：

1
2

pip3 install -U huggingface_hub hf_transfer -i https://pypi.tuna.tsinghua.edu.cn/simple
export HF_ENDPOINT=https://hf-mirror.com

　　同时下载模型到本地（需要具体看报错的是哪个模型，这里是wybxc/DocLayout-YOLO-DocStructBench-onnx）：

1

huggingface-cli download --resume-download wybxc/DocLayout-YOLO-DocStructBench-onnx --local-dir .

　　选择本地ollama模型作为翻译服务的话，需要配置环境变量：

1
2
3
4

# 本地ollama服务地址
export OLLAMA_HOST=http://127.0.0.1:11434 
# 选用phi4模型翻译
export OLLAMA_MODEL=phi4 

　　有个浏览器插件「沉浸式翻译 - Immersive Translate」同样是个比较好用的翻译工具。不过对于 PDF 文件，都有可能出现译文重叠的现象，需要二次编辑一下，或者将 PDF 格式转换为其他格式（eg：html，epub 等，相关 issue）。

VSC 辅助编程插件篇

Continue

　　 Continue 插件可结合本地的 ollama 使用 qwen2.5-coder:7b 模型，可辅助读/写代码，需打开以下设置：

• Continue: Enable Quick Actions
• Continue: Enable Tab Autocomplete
• Continue: Show Inline Tip

模型设置添加：

1
2
3
4
5

{
    "model": "qwen2.5-coder:7b",
    "title": "ollama-qwen2.5-coder",
    "provider": "ollama"
}

自动补全设置中 apiKey 保持为空字符串就行。

若需要使用远程部署的 ollama 服务，可以新增参数 "apiBase": "http://<my endpoint>:11434"。

Cline

　　 Cline 插件同样可结合本地的 ollama 使用，可辅助 CR 以及自动化优化修改代码。如果使用 vLLM 部署的 AI 服务，API Provider 选择 OpenAI Compatible，Bsae URL 填 http://ip:port/v1，API Key 随便填就行（eg: ollama），Model ID 则是模型名称（eg: deepseek-r1:14b）。

20250207 更新：

• 英译中模型推荐使用 qwen2.5:14b；
• 中文问答聊天模型推荐使用 deepseek-r1:14b，模型会输出详尽的思考过程；

后记

　　本地部署 LLM 的好处在于无数据泄漏问题，对于个人使用而言，轻量级的模型也差不多够用了，但即使已经轻量化了，本地运行大模型还是有点吃力，在 Shaun 的电脑上运行 phi4:14b 略显勉强（8 tok/s）。Mac 的内存和显存是共享的，后续如果买新的，有部署 LLM 的需求，最好把内存拉满，由于模型文件也相对较大，有条件的可以把 SSD 也拉满。希望后续大模型的推理能够进一步轻量化，效果也更好，真正实现人人都能使用。

　　自 GPT-3 出现以来，也就短短 4 年不到，从大规模的高性能 GPU 集群到单机部署，从胡言乱语到精准命中，各行各业都迎来了 LLM 的冲击，在可预见的未来，LLM 将深刻影响到每一个人，这种影响无关好坏，单纯只是时代的浪潮，LLM 将和操作系统，数据库一样，成为整个 IT 行业的基础设施，就 Shaun 而言，应该很难亲自动手去开发优化 LLM，能做的也就是尽可能的熟练使用。

参考资料

1、基于Ollama+AnythingLLM搭建本地私有知识库系统

2、ollama搭建本地个人知识库

前言

　　工作中偶尔会遇到一些特殊需求需要解决，这里记录一下。

需求篇

Mac 修改文件创建时间和修改时间

　　使用 setfile 命令：

修改创建日期：setfile -d "mm/dd/yy hh:mm:ss" filename

修改修改日期：setfile -m "mm/dd/yy hh:mm:ss" filename

同时修改 xxx.txt 文件两个时间为 2023-07-27 01:23:53：

setfile -d "07/27/2023 01:23:53" -m "07/27/2023 01:23:53" ./xxx.txt

Excel 修改创建时间

　　word 和 excel 本质上都是 zip 文件，可利用 openpyxl 修改 xlsx 文件元信息创建时间。对于 xls 文件，若文件有密码，需先去除密码，再将 xls 转换为 xlsx 文件，之后使用 openpyxl 修改时间。具体步骤如下：

1. 用 AppleScript 将 xls 转换为 xlsx

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26

   def run_applescript(script): """运行 AppleScript 脚本""" subprocess.run(["osascript", "-e", script]) def xls_to_xlsx(file_path="./xxx.xls"): """使用 AppleScript 修改 Excel 文件元数据""" applescript = f''' tell application "Microsoft Excel" -- 打开 .xls 文件 set inputFile to "{file_path}" -- 修改为你的文件路径 open inputFile -- 获取当前工作簿 set wb to active workbook -- 定义输出文件路径 set outputFile to "{file_path}x" -- 修改为你想保存的文件路径 -- 保存为 .xlsx 格式 save workbook as wb filename outputFile file format Excel XML file format -- 关闭工作簿 close wb saving no end tell ''' run_applescript(applescript)

2. 修改 xlsx 文件创建时间

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15

   import openpyxl def modify_excel_metadata(file_path = "./xxx.xlsx"): # 打开 Excel 文件 wb = openpyxl.load_workbook(file_path) # 获取元数据（properties） # properties = wb.properties # print(properties.__dict__) dt = datetime.strptime("2023-01-07 14:00:45", "%Y-%m-%d %H:%M:%S") dt -= timedelta(hours=8) wb.properties.creator = "" wb.properties.modified = dt wb.properties.created = dt wb.save("./xxx_tmp.xlsx")

3. 将 xlsx 转换为 xls 文件

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26

   def xlsx_to_xls(file_path="./xxx_tmp.xlsx"): """使用 AppleScript 修改 Excel 文件元数据""" applescript = f''' tell application "Microsoft Excel" -- 打开 .xls 文件 set inputFile to "{file_path}" -- 修改为你的文件路径 open inputFile -- 获取当前工作簿 set wb to active workbook -- 定义输出文件路径 set xlsFilePath to (inputFile as text) set xlsFilePath to text 1 thru -6 of xlsFilePath -- 去掉 ".xlsx" set xlsFilePath to xlsFilePath & ".xls" # log xlsFilePath -- 保存为 .xls 格式 save wb in xlsFilePath # save workbook as wb filename xlsFilePath file format Excel98to2004 file format with overwrite -- 关闭工作簿 close wb saving yes end tell ''' run_applescript(applescript)

JPG 修改创建时间

　　利用 pillow 和 piexif 修改 jpg 文件 exif 信息时间

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

from PIL import Image
import piexif

def modify_jpg_exif(img_file="./xxx.jpg", time_str = "2023:01:07 14:00:45"):
    im = Image.open(img_file)
    if "exif" not in im.info:
        return
    exif_dict = piexif.load(im.info["exif"])
    
    # for ifd in ("0th", "Exif", "GPS", "1st"):
    #     for tag in exif_dict[ifd]:
    #         print(ifd, tag, piexif.TAGS[ifd][tag], exif_dict[ifd][tag])
    
    del exif_dict["1st"]
    del exif_dict["thumbnail"]
    
    exif_dict["0th"][piexif.ImageIFD.DateTime] = time_str.encode()
    exif_dict["Exif"][piexif.ExifIFD.DateTimeOriginal] = time_str.encode()
    exif_dict["Exif"][piexif.ExifIFD.DateTimeDigitized] = time_str.encode()
    exif_bytes = piexif.dump(exif_dict)
    im.save("./xxx_m.jpg", exif=exif_bytes, quality='keep', subsampling='keep')

　　Pillow 保存 jpg 图片默认会同时保存 JFIF 和 EXIF 头，若需要去掉 JFIF 头，需修改 Pillow JpegEncode.c 文件源码:

1
2
3
4
5
6

if (context->xdpi > 0 && context->ydpi > 0) {
    context->cinfo.write_JFIF_header = TRUE;
    context->cinfo.density_unit = 1; /* dots per inch */
    context->cinfo.X_density = context->xdpi;
    context->cinfo.Y_density = context->ydpi;
}

修改为：

1

context->cinfo.write_JFIF_header = FALSE;

之后执行：python3 -m pip -v install . 从本地源码安装 Pillow。

PDF 修改创建时间

　　使用 pikepdf 修改 pdf 文件元信息时间

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

import pikepdf

def modify_pdf_metadata(file_path="./xxx.pdf", time_str = "20241203140045+08'00'"):
    # 打开 PDF 文件
    with pikepdf.open(file_path, allow_overwriting_input=True) as pdf:
        ## 获取 PDF 的元数据
        # metadata = pdf.docinfo
        # for key, value in metadata.items():
        #     print(f'{key}: {value}')
        
        # 修改元数据
        pdf.docinfo["/CreationDate"] = time_str
        pdf.docinfo["/ModDate"] = time_str
        # 保存文件
        pdf.save()

前言

　　最近业务调用方反馈接收到服务连接中断的错误（python requests 请求抛出异常 raise ConnectionError(err, request=request) \n ConnectionError: ('Connection aborted.', BadStatusLine("''",))），但从 golang 服务日志中看，服务应该是正常处理完成并返回了，且抛出异常的时间也基本和服务返回数据的时间一致，即表明在服务响应返回数据的那一刻，请求方同时抛出异常。

　　这个问题很奇怪，起初拿到一个 case 还无法稳定复现，最初怀疑是网络抖动问题，但后续一直会偶发性出现，直到拿到了一个能稳定复现的 case，深入跟踪排查后才发现与网络问题无关，是服务端框架应用设置不合理的问题。

问题篇

　　从网上搜索 python ConnectionError: ('Connection aborted.')，错误种类非常多，有网络问题，服务端问题（关闭连接，拒绝服务，响应错误等），客户端关闭连接，超时设置不合理，请求参数/协议错误等等，但若带上 BadStatusLine("''",) ，错误就相对比较明确了（BadStatusLine Error in using Python, Requests，Python Requests getting ('Connection aborted.', BadStatusLine("''",)) error），主要是由于收到了一个空响应（header/body），空响应可以明确是服务端返回的问题，一般可能有以下几个原因：1. 服务端反爬；2. 服务端超时（比如 nginx 默认 60s 超时）；3. 网络错误。

　　由于是内部服务，所以反爬策略是没有的，而反馈的 case 都带有明显的特征（请求数据量大，处理耗时长），没有网络抖动那种随机性，所以应该也不是网络问题，剩下的只能是超时问题，由于业务方在前置策略上已经识别该 case 数据量大，所以不经过 nginx 网关，直连服务请求，所以也不会有 nginx 超时问题，只能是服务端自己超时。于是直接在代码中查找 timeout 关键字，发现在服务启动时设置了 ReadTimeout 和 WriteTimeout，进一步深挖之后，才对 go 服务的超时有了浅显的认识。

超时篇

参考资料：1. 你真的了解 timeout 吗？，2. i/o timeout ， 希望你不要踩到这个net/http包的坑，3. net/http完全超时手册。

　　由于 HTTP 协议规范并未提及超时标准，而为保证服务稳定性，一般的 HTTP 服务请求都会设置超时时间，各 HTTP 服务端/客户端对于超时的理解大同小异，而这次的问题又起源与 go 服务，所以以 go 为例，分析一下超时。

客户端超时

　　客户端超时，即 GET/POST 请求超时，这个很好理解，就是客户端发送请求到客户端接收到服务器返回数据的时间，算是开发的一般性常识，控制参数一般也特别简单，就是一个 timeout，当然 go 服务客户端支持设置更精细化的超时时间，一般也没啥必要。当客户端感知到超时时，会正常发起 TCP 断开连接的“四次挥手”过程。

服务端超时

　　服务端超时，这才是引发问题的根本原因，go 服务端的超时，主要有两个参数，ReadTimeout 和 WriteTimeout，从上图可以看出，ReadTimeout 主要是设置服务端接收请求到读取客户端请求数据的时间（读请求的时间），WriteTimeout 是服务端处理请求数据以及返回数据的时间（写响应的时间）。GoFrame 框架的 ReadTimeout 默认值是 60s，在请求数据正常的情况下 ReadTimeout 也不可能超时，这次的问题主要出在 WriteTimeout，GoFrame 的默认值是 0s，代表不控制超时，但之前的开发者也同样设置为了 60s，导致服务端在处理大量数据时，发生了超时现象。

　　更深挖之后，才发现 WriteTimeout 的诡异之处，当 WriteTimeout 发生之后，服务端不会即时返回超时消息，而是需要等服务端真正处理完之后，返回数据时，才会返回一个空数据，即使服务端正常写入返回数据，但都会强制为空数据返回，导致请求客户端报错。这种表现，看起来就像是 WriteTimeout 不仅没有起到应有的作用，在错误设置的情况下，还会起到反作用，使服务响应错误。WriteTimeout 无法即时生效的问题，也同样有其他人反馈了：1. Diving into Go's HTTP server timeouts；2. net/http: Request context is not canceled when Server.WriteTimeout is reached。可能是网上反馈的人多了，go 官方推出了一个 TimeoutHandler，通过这个设置服务端超时，即可即时返回超时消息。仿照官方的 TimeoutHandler ，即可在 GoFrame 框架中也实现自己的超时中间件。

　　至于 WriteTimeout 为啥不起作用，个人猜测主要原因在于 go 服务每接收到一个请求，都是另开一个协程进行处理，而 goroutine 无法被强制 kill，只能自己退出，通常是要等到 goroutine 正常处理完之后才能返回数据，WriteTimeout 只是先强制写一个空数据占位，返回还是得等 goroutine 正常处理完。

　　所以正常的 go 服务，在使用类似于 TimeoutHandler 中间件的时候，也最好让 goroutine 尽可能快的退出，一种简单的方法是：1. 设置请求的 context 为 context.WithTimeout；2. 分步处理数据，每一步开始前都先检查请求传入的 context 是否已经超时；3. 若已经超时，则直接 return，不进行下一步处理，快速退出 goroutine。

后记

　　这次问题排查，碰到的最大障碍在于，前几次反馈的 case 难以复现，客户端请求报错和服务器返回的时间一致也不会让人往超时的角度去想，在拿到一个能稳定复现的 case 之后，才死马当活马医，先调一下超时参数试试。

　　关于 go 服务超时的文章，其实之前也看过，但没碰到具体问题，名词也就仅仅只是名词，很难理解背后的含义和其中的坑点，实践才能出真知 ╮(~▽~)╭。

附录

长连接超时

　　关于超时问题，也曾看到过有人碰到一个长链接服务的问题，现象是这样的：后端服务宕机之后，客户端可能需要很久才会感知到，原因在于 tcp 的超时重传机制，在 linux 中，默认会重传 tcp_retries2=15 次（即 16 次才会断开连接），而 TCP 最大超时时间为 TCP_RTO_MAX=2min，最小超时时间为 TCP_RTO_MIN=200ms。即在 linux 中，一个典型的 TCP 超时重传表现为：

所以客户端需要在 15 分钟之后才能感知到服务端不可用，如此，仅靠 TCP 自身的超时机制，很难发现服务端是否宕机/不可用，长链接不释放，进而可能导致客户端不可用且无感知，所以在长链接服务中，需要有其他的手段来保障服务稳定/可用性（eg：心跳探活）。

服务端 context canceled

Refer to: context canceled，谁是罪魁祸首

　　从官方的 net/http 包中可以知道，go 服务在接收请求时，会同时生成一个协程监控连接状态，当发现连接有问题（eg：客户端设置请求超时主动断开）时，会将该请求对应的 context cancel 掉，这时服务端如果再继续使用该 context 时，就会报错「context canceled」。当然，如果服务端发生错误，也同样会导致请求对应的 context cancel 掉。

　　服务端主动 cancel context 的好处在于可以快速释放资源，避免无效的请求继续执行（当然也得业务代码上主动去感知 context 是否 cancel，从而及时退出）；坏处在于，如果服务端需要上报这个请求发生的错误（一般在后置中间件中进行错误上报），这个时候上报错误的请求需要另外生成一个新的 context，绝不能直接使用现有的 context，因为已有的这个 context 已经 cancel 掉了，继续使用会导致上报错误的请求发送失败，达不到上报的目的。

前言

　　最近偶然在公司内网看到一篇文章「负载均衡算法vnswrr改进——从指定位置生成调度序列」。正好 Shaun 一直觉得调度类算法很有意思，就认真看了下，顺便写下自己的一些理解。

预备篇

　　通俗来讲负载均衡解决的是「在避免机器过载的前提下，多个请求如何分发到多台机器上」的问题，本质上是一个分布式任务调度的问题，在机器性能相同的情况下，最简单的策略就是轮询，多个机器依次轮流处理请求。Nginx 官方的 SWRR 算法解决的是「在机器性能不同的情况下，如何使请求分布更均匀，更平滑，避免短时间大量请求造成局部热点」的问题。

SWRR篇

　　在 SWRR 算法中，有两个权重，一个是初始实际权重（effective weight， ew），一个是算法迭代过程中的当前权重（current weight，cw），在负载均衡过程中，每次请求分发都选择当前权重最大的机器，同时更新每台机器的当前权重，当前权重更新策略如下：

0. 若设定 n 台机器各自的初始权重为 $(ew_1,ew_2,...,ew_n)$，同时 $ew_1 \le ew_2 \le ... \le ew_n$ ，且 $W_{total}=\sum_{i=1}^n ew_i$ ；

1. 第一个请求来时，n 台机器各自的当前权重 $cw_i=ew_i,　1 \le i \le n$ ，由于此时 $cw_{max}=\max(cw_i)=cw_n$ ，则请求分发给第 n 台机器处理，同时更新机器各自的当前权重 $cw_1=cw_1+ew_1, cw_2=cw_2+ew_2,...,cw_{n-1}=cw_{n-1}+ew_{n-1},cw_n=cw_n+ew_n-W_{total}$，记为 $(2*ew_1,2*ew_2,...,2*ew_{n-1},2*ew_n-W_{total})$ ；

2. 第二个请求来时，此时 n 台机器的各自权重为 $(2*ew_1,2*ew_2,...,2*ew_{n-1},2*ew_n-W_{total})$ ，选取权重值对应的机器进行处理，假设为第 n-1 台，则更新后权重为 $(3*ew_1,3*ew_2,...,3*ew_{n-1}-W_{total},3*ew_n-W_{total})$ ；

3. 第 $W_{total}$ 个请求来时，此时 n 台机器的各自权重应该为 $$(W_{total}*ew_1-m_1*W_{total},W_{total}*ew_2-m_2*W_{total},...,W_{total}*ew_{n-1}-m_{n-1}*W_{total},W_{total}*ew_n-m_n*W_{total}) \\ \text{s.t.} \quad \sum_{i=1}^n m_i=W_{total}-1 \\ \quad 0 <= m_i <= ew_i$$ 由于每次调度都是权重最大值减权重和，重新分配权重后权重和无变化，所以理论上此时除第 k 台机器外，每台机器的权重都为 0，第 k 台机器的权重为 $W_{total}$ ，所以这次调度处理之后，每台机器的权重又会重新回到初始权重。

VNSWRR 篇

　　VNSWRR 算法是阿里针对 Nginx 官方的 SWRR 算法实际运行中对于部分场景下（瞬时流量大，权重更新等）均衡效果不太理想的改进算法，其最大的改进点在于预生成调度序列，以空间换时间减少调度时间，同时在权重更新后随机选取调度序列的起点，使初次请求就调度在不同的机器上，减少高权重机器的局部热点问题。具体流程如下：

1. 首先使用 SWRR 算法生成前 n 个调度序列；
2. 再随机选取一个位置作为调度起点，后续的请求依次从调度序列中选取；
3. 若调度序列用完，则继续用 SWRR 算法生成后 n 个调度序列；
4. 如此循环，直到调度序列的长度为 $W_{total}$，即一个周期内的全部调度序列，用完后，从头开始调度即可；
5. 若有权重更新，则从 1 开始重新生成调度序列；

正文

　　从上面的逻辑中，可看出 SWRR 算法调度序列是以 $W_{total}$ 为周期的一个循环序列，只需要知道一个周期内的调度序列，就可以推算出后续的调度机器（除非权重有变更或者有机器增删）。计算一个周期内的调度序列也比较简单，取当前调度权重中最大值对应机器，同时更新每台机器的当前权重，作为下次调度的权重，简而言之，就是从上次调度结果推出下次调度结果，是一个递推式。那有没有办法不从上次结果推下次结果，直接计算当前的调度结果，简化 VNSWRR 的第一步每次都从头开始预生成前 n 个调度序列，直接从任意位置开始生成调度序列，内网中这篇文章就给出了一个看似“可行的”解决方案，直接计算第 q 个请求的调度结果，具体方案如下：

在 SWRR 算法中，第 q 个请求时，全部机器的当前权重序列应该为 $$(q*ew_1-m_1*W_{total},q*ew_2-m_2*W_{total},...,q*ew_{n-1}-m_{n-1}*W_{total},q*ew_n-m_n*W_{total}) \\ \text{s.t.} \quad \sum_{i=1}^n m_i=q-1 \\ \quad 0 <= m_i <= ew_i$$ 即权重序列中共减去了 $q-1$ 个 $W_{total}$ ，平均上 $m_i=ew_i/W_{total}*(q-1)$，区分 $m_i$ 的整数部分 $mz_i$ 和小数部分 $mx_i$，$\sum_{i=1}^n m z_i$ 代表减去的 $W_{total}$ 个数，计算差值 $d=q-1-\sum_{i=1}^n mz_i$，即还剩 d 个 $W_{total}$ 待减，对小数部分 $mx_i$ 从大到小排序，取前 d 个对应的机器再减 $W_{total}$，即可得到第 q 个请求时的当前权重序列，取最大权重对应的机器即为调度结果，后续调度结果可通过递推式得出。

　　初次看到这个方案的时候，就想动手实现一下，因为思路也比较清晰简单，实现完之后，简单测试一下，也确实没啥问题，后面再深度测试了一下，就发现该方案确实有点小小的问题，在大部分情况下，该方案确实能得到很正确的结果，但还是存在一些错误结果，就因为有少量错误结果，所以该方案不要在生产环境下应用。该方案错在了将 $q*ew_i$ 看成最后一个整体进行处理排序，忽略了分步执行结果，导致小部分场景下的错误排序结果，进而生成错误调度权重，调度错误。

　　现在再回到初始问题「如何生成 SWRR 算法中指定轮次的调度结果？」，抽象来看，该问题是个数学问题「如何从数列的递推式计算数列求通项公式」， 但 SWRR 的递推式相对复杂，中间还有取最大值这个不稳定变量，实际很难得到通项公式，直接计算指定调度解果，Shaun 问了 ChatGPT，也自己想了很久，搜了很久，但都没有答案，内网中的这个方案算是最接近的一个答案。

后记

　　在内网中看到这个方案的思路很有意思，将整数和小数部分拆开，再单独对小数部分排序，所以就自己测试了一下，顺便学习了下负载均衡 SWRR 算法，虽然问题依旧还在，但总归是有点收获。

附录

　　附代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111

import random


def ouput_schedule(rs_arr, schedule_num):
    all_rs_weight_str = ";\t".join(["rs:%s,cw:%s" % (rs["rs_name"], rs["cw"]) for rs in rs_arr])
    schedule_rs = max(rs_arr, key=lambda x:x["cw"])
    print("%s:\t%s\t===>\trs:%s,cw:%s" % (schedule_num, all_rs_weight_str, schedule_rs["rs_name"], schedule_rs["cw"]))

    return schedule_rs

def swrr(rs_arr, weight_total):
    schedule_rs = rs_arr[0]
    max_weight = schedule_rs["cw"]
    for rs in rs_arr:
        if rs["cw"] > max_weight:
            schedule_rs = rs
            max_weight = rs["cw"]

        rs["cw"] += rs["ew"]
    
    schedule_rs["cw"] -= weight_total

    return schedule_rs

def swrr_test():
    real_servers = [{"rs_name": chr(i+64), "ew": i, "cw": i} for i in range(1, 6)]
    weight_total = sum([rs["ew"] for rs in real_servers])
    schedule_count = weight_total
    swrr_seq = []
    for i in range(1, schedule_count+1):
        ouput_schedule(real_servers, i)
        schedule_rs = swrr(real_servers, weight_total)

        swrr_seq.append(schedule_rs["rs_name"])
    
    print(swrr_seq)

# swrr_test()
# print("---------")

def swrr_n(rs_arr, weight_total, schedule_num):
    ms = [(rs["ew"] / float(weight_total)) * (schedule_num-1) for rs in rs_arr]
    mzs = [int(m) for m in ms]
    mxs = [(i, m-int(m)) for i, m in enumerate(ms)]
    mxs = sorted(mxs, key=lambda x:x[1], reverse=True)
    for i, rs in enumerate(rs_arr):
        rs["cw"] = schedule_num * rs["ew"]
        rs["cw"] -= mzs[i] * weight_total

    d = (schedule_num-1) - sum(mzs)
    for i in range(d):
        rs_arr[mxs[i][0]]["cw"] -= weight_total

    schedule_rs = ouput_schedule(rs_arr, schedule_num)

    return schedule_rs

def swrr_n_test():
    real_servers = [{"rs_name": chr(i+64), "ew": i, "cw": i} for i in range(1, 6)]
    weight_total = sum([rs["ew"] for rs in real_servers])

    schedule_rs_seq = []
    for i in range(1, weight_total+1):
        schedule_rs = swrr_n(real_servers, weight_total, i)

        schedule_rs_seq.append(schedule_rs["rs_name"])
    # swrr_n(real_servers, weight_total, 9) # err schedule rs
    print(schedule_rs_seq)

# swrr_n_test()

def vnswrr_preschedule(rs_arr, weight_total, N, schedule_rs_seq):
    for i in range(1, N+1):
        schedule_rs = swrr(rs_arr, weight_total)
        if len(schedule_rs_seq) >= weight_total:
            break
        schedule_rs_seq.append(schedule_rs)

def vnswrr(rs_arr, rs_count, weight_total, prev_schedule_idx, schedule_rs_seq):
    N = min(rs_count, weight_total)
    
    schedule_idx = prev_schedule_idx + 1
    schedule_idx %= weight_total
    
    if schedule_idx >= len(schedule_rs_seq)-1:
        vnswrr_preschedule(rs_arr, weight_total, N, schedule_rs_seq)
    
    return schedule_idx

def vnswrr_test():
    all_schedule_rs_seq = []
    real_servers = [{"rs_name": chr(i+64), "ew": i, "cw": i} for i in range(1, 6)]
    rs_count = len(real_servers)
    weight_total = sum([rs["ew"] for rs in real_servers])

    N = min(rs_count, weight_total)
    schedule_rs_seq = []
    # 预生成调度序列
    vnswrr_preschedule(real_servers, weight_total, N, schedule_rs_seq)
    # 随机取调度结果
    prev_schedule_idx = random.randint(0, N-1)-1

    for i in range(1, 2*weight_total+1):
        schedule_idx = vnswrr(real_servers, rs_count, weight_total, prev_schedule_idx, schedule_rs_seq)
        all_schedule_rs_seq.append(schedule_rs_seq[schedule_idx]["rs_name"])
        prev_schedule_idx = schedule_idx
    
    print([rs["rs_name"] for rs in schedule_rs_seq])
    print(all_schedule_rs_seq)

vnswrr_test()

参考资料

1、QPS 提升60%，揭秘阿里巴巴轻量级开源 Web 服务器 Tengine 负载均衡算法

2、Nginx SWRR 算法解读

前言

　　最近发现一个 GoFrame 服务即使空载 CPU 使用率也很高，每次接受请求后资源没有被释放，一直累积，直到达到报警阈值，人工介入重启服务，于是压测排查了一下。

问题篇

　　先新增代码启动 go 自带的 pprof 服务器：

1
2
3
4
5
6
7
8
9
10
11
12

package main

import (
	"net/http"
	_ "net/http/pprof"
)

func Pprof(pprof_port string) {
	go func(pprof_port string) {
		http.ListenAndServe("0.0.0.0:"+pprof_port, nil)
	}(pprof_port)
}

压测以及 profile 命令：

1
2
3
4
5
6
7
8
9
10
11
12

# 压测命令
wrk -t8 -c1000 -d60s --latency --timeout 10s -s post_script.lua http://host:[srv_port]/post

# profile 整体分析
go tool pprof -http=:8081 http://host:[pprof_port]/debug/pprof/profile?seconds=30

# 查看函数堆栈调用
curl http://host:[pprof_port]/debug/pprof/trace?seconds=30 > ./pprof/trace01
go tool trace -http=:8081 ./pprof/trace01

# 查看内存堆栈
go tool pprof -http=:8081 http://host:[pprof_port]/debug/pprof/heap?seconds=30

　　在压测 30 次后，即使服务空载 CPU 也被打满了，查看服务此时的 profile，发现 goroutine 的数目到了百万级别，查看 cpu 堆栈发现集中调用在 gtimer 上，但遍寻服务代码，没有直接用到 GoFrame 的定时器，问题出在哪也还是没想太明白。吃完饭后偶然灵光一现，既然 CPU 看不出啥，那再看看内存，查看内存发现，内存对象最多的是 glog.Logger，看代码也正好有对应的对象，可算是找到问题真正的元凶了。

　　log 对象一般都是全生命周期的，不主动销毁就会一直伴随着服务运行，所以 log 对象一般都是程序启动时初始化一次，后续调用，都是用这一个对象实例。而这次这个问题就是因为在代码中用 glog 记录了数据库执行日志，每次请求都会重新生成一个 glog 对象，又没有主动释放造成的。

　　知道问题的真正所在，解决问题就相对很简单了，只在程序启动时初始化一个 glog 对象，后续打印日志就用这一个实例，其实更好的方式是生产环境不打印数据库日志，毕竟影响性能。

后记

　　CPU 资源的占用往往伴随着内存资源的占用，当从调用堆栈以及线程资源上看不出问题的时候，可以转过头来看看内存堆栈，毕竟内存堆栈更能指示有问题的对象出在哪，知道内存对象是谁，也相当于提供了排查问题代码的方向。

附录

　　在排查过程中发现 goroutine 数目异常的高，于是想限制一下 goroutine 数目，在网上搜索的时候发现当用容器部署 go 服务时，go 默认最大的 goroutine 数目为宿主机 cpu 核数，而不是容器的 cpu 核数，从而并发时 goroutine 数目可能比容器 cpu 核数高很多，造成资源争抢，导致并发性能下降，可以通过设置环境变量 GOMAXPROCS 指定 goroutine 最大数目，也可以使用 go.uber.org/automaxprocs 库自动修正最大核数为容器 cpu 核数。

自适应设置 GOMAXPROCS 上下限代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

package main

import (
	_ "go.uber.org/automaxprocs"

	"runtime"
)

func main() {
	procsNum := runtime.GOMAXPROCS(-1)
	if procsNum < 4 {
		procsNum = 4
	} else if procsNum > 16 {
		procsNum = 16
	}

	runtime.GOMAXPROCS(procsNum)

	// todo something...

}

python 内存泄漏排查

※注：python 的默认参数是全局变量，若默认参数为一个引用类型（eg：字典对象），且函数中会对该参数进行写操作，就极有可能发生内存泄漏，所以 python 默认参数最好是值类型。

方法一是线上程序直接排查，通过 pyrasite 和 guppy 直接对应 python 程序：

step1：绑定 python 程序 pid，开启 pyrasite shell 窗口，执行 pyrasite-shell <pid>；

step2：使用 guppy 查看 python 程序内存情况，

1 2 3

>>> from guppy import hpy >>> h = hpy() >>> h.heap()

step3：间隔一定时间后，再次使用 h.heap()，对比两次内存变化

该方法一般只能粗略查看内存泄露的数据对象，可能无法精确定位到指定位置，这时需要用方法二，手动插入代码查看程序运行日志：

Python标准库的gc、sys模块提供了检测的能力

1 2 3 4 5 6

import gc import sys gc.get_objects() # 返回一个收集器所跟踪的所有对象的列表 gc.get_referrers(*objs) # 返回直接引用任意一个 ojbs 的对象列表 sys.getsizeof() # 返回对象的大小（以字节为单位）。只计算直接分配给对象的内存消耗，不计算它所引用的对象的内存消耗。

基于这些函数，先把进程中所有的对象引用拿到，得到对象大小，然后从大到小排序，打印出来，代码如下：

1 2 3 4 5 6 7 8 9 10 11

import gc import sys def show_memory(): print("*" * 60) objects_list = [] for obj in gc.get_objects(): size = sys.getsizeof(obj) objects_list.append((obj, size)) for obj, size in sorted(objects_list, key=lambda x: x[1], reverse=True)[:10]: print(f"OBJ: {id(obj)}, TYPE: {type(obj)} SIZE: {size/1024/1024:.2f}MB {str(obj)[:100]}")

找到内存占用稳定增长的对象，调用 gc.get_referrers(*objs)，查看该对象的引用信息，即可快速定位泄漏位置

该方法更加灵活精确，不好的地方是有侵入性，需要修改代码后重新上线，同时获取这些信息并打印，对性能有一定的影响，排查完之后，需要将该段代码下线。

参考资料

1、python内存泄露问题定位：附带解决pyrasite timed out

2、技术 · 一次Python程序内存泄露故障的排查过程