普贤教程 › 生态包

并发与常见模式

限流、重试、缓存、定时、并发原语。这一组包解决的是「服务跑久了才会出问题」的那类事——它们的共同点是状态显式传递，没有隐藏全局变量。

限流：令牌桶

import rate

var bucket = rt_new(10, 2.0)          # 容量 10，每秒补 2 个
print(rt_allow(bucket, 1))
# [{cap: 10, last: 1, rate: 2.0, tokens: 9}, true]

rt_allow 返回的是「新桶 + 布尔」的二元组不是 bool。因为它改写桶状态（扣掉令牌），返回值是新状态——必须接住： var r = rt_allow(bucket, 1) → bucket = r[0] → var ok = r[1]。只写 if rt_allow(bucket, 1): 会永远为真（非空 list 为真）。

def allow(bucket, n):
    var r = rt_allow(bucket, n)
    return [r[0], r[1]]

var b = rt_new(100, 10.0)
var ok = false
var pair = allow(b, 1)
b = pair[0]
ok = pair[1]

重试：带退避

import retry

var cfg = {"attempts": 5, "base_ms": 100, "strategy": "exp", "max_ms": 5000}
print(retry_backoff_series(5, 100, "exp"))     # 100/200/400/800/1600（含抖动另算）
print(retry_max_delay(5, 100, "exp"))

var r = retry_call(fetch_fn, null, cfg)        # 可重试错误才重，不可重试立刻返回

只对「可重试」的错误重试retry_is_retryable(err) 判断：连接超时、5xx 可重试；400、参数错误、签名失败不可重试。把不可重试的错误也重试，只会把一次失败放大成五次。

缓存：TTL + LRU

import cache

var c = ca_new(100)                    # 最多 100 条
ca_set(c, "k1", "v1", 0)               # ttl 0 = 不过期
print(ca_get(c, "k1").ok())            # v1
print(ca_has(c, "k1", 5000))           # 带 now 判断是否过期

ca_set(c, "k2", "v2", 1, 1000)         # 1 秒过期
print(ca_has(c, "k2", 5000))           # false

时间是显式传入的（默认 now_sec()），所以单测里可以把「现在」设成任意值，不用 sleep。过期是惰性清理：取的时候才判。

定时任务

import sched

var s = sched_new()
sched_add(s, "cleanup", 60000, cleanup_fn)     # 每 60 秒
print(sched_counts(s))
sched_run(s, 3600000)                          # 推进 1 小时（虚拟时钟）

sched 是「时间片轮询 + 显式时钟」，不是 systemd timer：它不会自己起线程，你在主循环里推它。真正的常驻定时任务用 systemd timer 或 cron，sched 适合模拟与单测。

并发原语

包	作用	注意
`semaphore`	计数限流	`sp_acquire` 拿不到会等；有 `sp_try_acquire` 不阻塞版本
`workerpool`	固定 worker 池跑任务	`wp_run` 提交、`wp_close` 收尾
`concurrent_map`	加锁的 dict	只在多协程共享时用；单协程纯属浪费
`actor`	消息循环 + 状态	状态只在一个线程里改，天然免锁
`counter`	计数（Counter 心智）	`ct_most_common` 排行
`dist`	随机分布采样	压测造流量、模拟抖动

actor：把状态关进一个线程

import actor

# 处理函数签名：(state, body) → 新 state
def on_inc(state, body):
    var n = state["n"] + 1
    return {"n": n}

def main():
    var a = ac_new({"inc": on_inc}, {"n": 0})
    print(ac_send(a, "inc", null))     # Ok(true)
    print(ac_status(a))                # running
    ac_stop(a)

实测输出就是 Ok(true) 与 running。actor 在自己的线程里串行处理消息，所以 state 不需要加锁——这是它比 concurrent_map 更省心的原因。

处理函数里改 state 要写成多行块单表达式赋值当表达式用会语法错；写成 var n = … + return {…} 最稳。

counter：一次算完频次

import counter
var c = ct_count(["a", "b", "b", "c", "c", "c"])
print(c)                    # {a: 1, b: 2, c: 3}
print(ct_most_common(c))    # [[c, 3], [b, 2], [a, 1]]

函数式组合

import functools
import itertools

print(it_combinations([1, 2, 3], 2))   # [[1, 2], [1, 3], [2, 3]]
print(it_permutations([1, 2, 3], 2))   # [[1, 2], [1, 3], [2, 1], [2, 3], [3, 1], [3, 2]]
print(it_chain([1, 2], [3, 4]))        # [1, 2, 3, 4]
print(it_unique([1, 1, 2]))            # [1, 2]

functools 提供 fn_partial / fn_compose / fn_memo / fn_pipe。fn_memo 的记忆化对纯函数有用，对依赖时间/随机的函数是灾难——缓存的是第一次的结果。

这一组包共同的坑

状态显式传：没有隐藏全局。多请求共享一个「桶」时，桶得存在模块级变量或数据库里。
「返回新状态」的函数必须接住：rt_allow、xsx_add、pd_add_page 都是这个模式。
多进程不共享内存：Ma 的 px 脚本是「每请求一个进程」，进程内限流只能防单进程突发——真正的全局限流要放 Redis（INCR + EXPIRE）。