concurrency

README

concurrency

并发与流控原语——GoSafe safe-go/协程池/重试/键控闸门/令牌桶/滑动窗口统计/自适应熔断/CPU 自适应降载

🌐 项目主页：https://oekit.github.io/ · 名称由来：Open Easy Kit（开源 · 易用 · 工具集合） 🤖 AI Agent：本仓规范与硬规则见 AGENTS.md；机器可读索引见 llms.txt（接口签名以源码为准）。

安装

go get github.com/oekit/concurrency@v0.5.0

这个仓库做什么

concurrency 是并发与流控原语集：把 Go 并发开发里最容易出事的部分—— 裸 goroutine 的 panic、无界重试、无界并发、限流、并发去重——封装成开箱即用且经过生产验证的原语。

核心能力

包	做什么
async	GoSafe 系列 safe-go：GoSafe/GoSafeWithErr/GoWithContext/GoSafeWithContext/GoSafeWithErrChannel/RunSafe，以及带任务归属标签的 RunSafeLabeled/GoSafeLabeled（panic 日志 task=…，可按任务聚合告警）。panic 自动 recover + 可注入 panic 上报。全矩阵禁止裸 go func 的替代品
gopool	自适应协程池：惰性拉起 worker、队列满回退直接 go（不丢任务）、空闲按 WorkerMaxAge 老化退出、Close(ctx) 优雅排空。高频突发任务的并发数软上限（v0.2.0 吸收自 cloudwego/gopkg）
retry	Do(ctx, cfg, fn) 重试/轮询：次数、固定/线性/指数退避（Exponential + 可选 Jitter 抖动、MaxDelay 封顶）、可重试判定
loop	周期任务循环原语：ctx 感知 + 单轮 panic 收编 + 忙时丢 tick + 排水（Wait/Done）。过期清理/心跳/快照落盘等手搓循环的收编件
syncx	键控闸门（同一 key 串行）、扇出控制
ratelimit	令牌桶/滑动窗口限流 + 按 key 隔离的 KeyedLimiter（泛型 key、容量护栏 WithMaxEntries、空闲回收、AllowWithRetryAfter 不消耗探测、Replace 速率热更）
window	滑动窗口统计容器（v0.3.0 移植自 go-zero collection.RollingWindow）：泛型桶 BucketInterface + 默认 Bucket{Sum, Count}、惰性重置、Reduce 聚合、IgnoreCurrentBucket；时间源可注入 WithWindowClock——熔断/降载的数据源
breaker	自适应熔断（v0.3.0 移植自 go-zero googleBreaker）：Google SRE 概率式丢弃、无状态机，40×250ms 窗口 + 强制探活自愈，panic 记失败后 re-panic；ErrCircuitOpen 全库规范哨兵（httpx/nns-endpoint 别名共用）——下游保护的执行件
shedding	CPU 自适应降载（v0.3.0 移植自 go-zero adaptiveshedder）：maxPass×minRt 的 Little's Law 并发上限、CPU 过载因子夹 [0.1,1]、avgFlying 滞后 EMA、1s 冷却期；CPU 采样纯标准库（/proc/stat + cgroup v1/v2 配额感知），非 linux 自动禁用全放行；ErrServiceOverloaded 哨兵——本机过载保护的入口件
singleflight	并发去重（同 key 合并为一次调用）
group	并发分组编排
progress	多阶段进度追踪

快速使用

import (
    "github.com/oekit/concurrency/async"
    "github.com/oekit/concurrency/gopool"
    "github.com/oekit/concurrency/retry"
)

// 替代裸 go func：panic 被 recover，可注入 panic 上报
async.GoSafe(func() { serve() })

// 高频突发任务走池：worker 数软上限，panic 汇入 logx
gopool.SetPanicHandler(func(ctx context.Context, r any, stack []byte) {
    logx.Error("gopool panic", "error", r, "stack", string(stack))
})
gopool.Go(func() { process(item) })

p := gopool.NewPool("ingest", gopool.PoolOption{MaxIdleWorkers: 64})
p.CtxGo(ctx, func() { handle(req) })
defer p.Close(ctx) // 优雅排空

// 带重试的调用
err := retry.Do(ctx, retry.Config{}, func(attempt int) error {
    return callDownstream()
})

边界

• 需要日志记录 panic → 配合 logx 的 SetPanicHandler
• 需要进程内缓存/去重合并的高级组合 → data
• 本仓为 L0 基础件：零矩阵内依赖（第三方仅 golang.org/x/time）

矩阵总览（23 活跃模块，自动生成）

模块	内容	矩阵内依赖
pkg	微工具集 21 合 1（errors/bloomx/contextx/container/validator/stringx/slicex/enumx/convert/encoding/jsonrepair/httpjson/mathx/pathx/pagination/ptrx/timex/uidgen/cache/bufiox/strmap）；cache（v0.4.0 起）含缓存接口 CacheAside + BatchGetter 批量能力（MGet）+ LocalCache/SingleFlightCache + cachetest 契约 + Checker；container（v0.9.0 起，收编原 collectionx）泛型容器 Set/Ring/Queue/SafeMap + Map（sync.Map 背书）/Slice（并发安全）；timex/stringx（v0.7.0 起）时间段边界 BeginningOf/EndOf 七对（分/时/日/周/月/季/年，周起始显式参数）+ trie 多模式替换器 Replacer（最长匹配）/ NewMask 敏感词遮罩；singleflighttest（v0.8.0 起）singleflight 共享契约测试套件（internal 与 concurrency 双侧引用）	—
concurrency	并发与流控原语（async/gopool/group/singleflight/progress/syncx/ratelimit/retry）+ window 滑动窗口 / breaker 自适应熔断（规范哨兵 ErrCircuitOpen）/ shedding CPU 自适应降载 / batch 攒批执行器（满条数/满字节/窗口去重）/ timewheel 单级时间轮	pkg
logx	统一日志门面（多后端可替换 + Sink + caller 归属）+ 优雅关停；桥接子包 hlog/hertzlog/kitexlog 为嵌套模块按需引入	—
core	应用基建件（governance 治理 schema / config 配置加载 + default/env/range/options tag 校验 / lifecycle 契约 / network IP 工具 / version）	—
metrics	Prometheus 门面 + 弹性指标族	core
tracing	OpenTelemetry 追踪 + trace_id 真相源	logx
registry	注册发现核心 + Nacos 实现 + Nacos 配置源 + sdbridge	core、logx
data	数据访问门面——复刻 ekit data/ 能力面：cache/rediscache/crud/distlock/gormx/redisx/timertask/cachetest 八子包 re-export（类型同一，实现住 pkg 与各子仓）+ cached 缓存读写编排（写后失效/事务提交后失效/查询缓存版本号失效/GetByIDs 批量点读 MGet 快路径/bloom 防穿透）+ twolevel 两级缓存与跨实例失效广播（v0.7.0 起本仓唯二非 re-export 实现）；all/ 聚合模块已废弃 retract	concurrency、gormx、logx、pkg、redisx、timertask
gormx	SQL/GORM 全家——连接池增强、框架无关 CRUD Service、仓储（原 oekit/gorm，v0.3.0 更名）	logx、pkg
redisx	Redis 全家——客户端封装（根包）、分布式缓存 cache/、分布式锁 distlock、分布式限流 ratelimit（Lua 三态 + 故障本地兜底）	concurrency、logx、pkg
etcdx	etcd 全家——客户端封装（根包：生产默认值/Health 探活/Close(ctx) 幂等）、registry 注册发现实现（SPI 对接 registry/nacos 链路：Lease KeepAlive 注册 + 丢约指数退避自动重注册、prefix Watch 订阅推送、ConfigStore KV 配置监听）、distlock 分布式锁（Session+concurrency.Mutex 公平排队，接口与 redisx/distlock.Locker 同形可互换，可选无 etcd 降级放行）	concurrency、core、logx、registry
httpx	HTTP 客户端 + 负载均衡——服务发现/熔断（规范哨兵 ErrCircuitOpen）/重试/指标/OTel 透传/一致性哈希（权重感知虚拟副本）	concurrency、core、logx、metrics、registry、tracing
network	轻量网络客户端——WebSocket/NATS/MQ 抽象、连接与流工具	concurrency、logx、tracing
timertask	分片定时任务（FNV-1a、心跳续约）	concurrency、logx
middleware	Hertz 自研标准中间件栈 + contrib 桥接 + hmw + SheddingMiddleware 自适应降载	concurrency、core、logx、metrics、pkg、registry、tracing
k8s	Kubernetes 客户端封装——RESTConfig/Clientset/DynamicClient 构建（显式 kubeconfig → in-cluster 自动 → KUBECONFIG/~/.kube/config 逐级回退，QPS/Burst 限流默认 50/100）+ leaderelection 子包 Lease 选主简化封装（OnStartedLeading/OnStoppedLeading/OnNewLeader 回调，主备单活，仅要求 Lease CRUD 权限）	logx
nns	NNS 服务治理客户端——provider 无关发现/配置/实例缓存	concurrency、core、logx、registry
nns-endpoint	NNS 端点工厂——per-service 熔断、换实例重试、gRPC 池	concurrency、core、httpx、logx、metrics、nns、registry
bootstrap	应用生命周期引导——bootstrap.Run 骨架 + healthz	concurrency、logx
hertz	Hertz 运行时集成——日志桥接、生命周期、SSE/health/metrics、适配器、OTel、crudh/validatorh/versionh（原 handler 微库收编）	bootstrap、concurrency、core、gormx、logx、metrics、middleware、nns、pkg、registry、tracing
kitex	Kitex 生态——日志桥接、治理热更、生命周期、BizError 桥接、kitexotel、istio（xDS 动态熔断，原 istio 微库收编）	bootstrap、concurrency、core、logx、metrics、nns、pkg、registry
workflow	组合式流程编排——串行/并行/异步/条件分支 + Hook AOP	concurrency、logx
kitex-examples	Kitex echo 示例——client/server + IDL + 治理配置	bootstrap、concurrency、core、kitex、logx、metrics、nns、pkg、registry、tracing

依赖严格自底向上；微库已合并为 pkg，旧微库仓已删除。

本仓在矩阵中的位置与依赖见上表「矩阵内依赖」列。历史沿革：源自 ekit 单体库 （git.enjoye.top/enjoydream/ekit v0.34.0，冻结维护），2026-10 拆分为独立 module。

License

MIT — 见 LICENSE