px普贤教程PuXian
普贤教程 › 生态包

Redis 实战

registry/redis 是用纯 PuXian 写的 RESP 协议客户端：不依赖 C 扩展、不依赖 redis-cli，编译后就是一个静态二进制里的客户端。本页的完整对话，是拿本机一个 40 行 PuXian 写的 RESP 服务器实测出来的。

安装

bash
pxpkg init
pxpkg add redis@0.2.0
pxpkg install
px
import redis

最小可用：连接、写入、读回

px
import redis

def main():
    var c = redis_connect("127.0.0.1", 6379, {"timeout_ms": 2000})
    if c.is_err():
        print("连不上：" + str(c.err()))
        return
    var conn = c.ok()

    print(redis_set(conn, "user:1", "张三"))     # Ok(true)
    print(redis_get(conn, "user:1"))             # Ok(张三)
    print(redis_incr(conn, "pv"))                # Ok(1)
    print(redis_close(conn))                     # Ok(true)
返回值一律是 Ok / Err库里没有隐藏异常：连接失败、命令报错都是 Err(msg)，程序继续跑。这是 PuXian 没有 try/catch 时的标准写法——把错误当值往后传。

实测对话（完整输出）

下面这段输出不是手写的，是编译产物 ./build/rdclient 对本地 RESP 服务器跑出来的：

text
connect => true
ping => Ok(PONG)
set => Ok(true)
get => Ok(张三)
get 不存在 => Ok(null)
exists => Ok(1)
incr => Ok(1)
incr => Ok(2)
get pv => Ok(2)
expire => Ok(1)
ttl => Ok(60)
hset => Ok(1)
hget => Ok(李四)
mset => Ok(OK)
mget => Ok([v1, v2])
del => Ok(1)
get k1 删后 => Ok(null)
close => Ok(true)
连不上时 ok => false
连不上 err => redis: connect 127.0.0.1:16380 failed: net: 连接 127.0.0.1:16380 失败 (111)

四个必须记住的返回形态

调用返回说明
redis_set(conn, k, v)Ok(true)不是 Ok("OK")，判成功要判 is_ok()
redis_get(conn, k)Ok(str) / Ok(null)键不存在是「成功的 null」，不是 Err
redis_mset(conn, pairs)Ok(...)pairs 是 [["k1","v1"],["k2","v2"]]，不是 dict
redis_del(conn, keys)Ok(int)keys 必须是 list：["k1"]，传字符串会逐字符拼命令
传 dict 给 redis_mset 会报 字典索引键必须是字符串它的实现是 pairs[i][0]，对 dict 取下标拿到的是键名——错在类型，不在库。

各数据类型的接口

类别函数
连接redis_connect(host, port, opts) · redis_ping · redis_select · redis_close
字符串redis_set · redis_get · redis_getset · redis_append · redis_strlen · redis_set_ex · redis_set_nx · redis_set_xx
计数redis_incr · redis_incrby · redis_decr
批量redis_mset(pairs) · redis_mget(keys)
键管理redis_del · redis_exists · redis_expire · redis_ttl · redis_type · redis_keys · redis_scan
Hashredis_hset · redis_hget · redis_hdel · redis_hexists · redis_hgetall · redis_hkeys · redis_hvals · redis_hlen
Listredis_lpush · redis_rpush · redis_lpop · redis_rpop · redis_lrange · redis_llen
Setredis_sadd · redis_srem · redis_smembers · redis_sismember · redis_scard
ZSetredis_zadd · redis_zrange · redis_zscore · redis_zrem
发布订阅redis_publish · redis_subscribe · redis_unsubscribe
事务redis_multi · redis_exec · redis_discard
任意命令redis_cmd(conn, ["SET", "k", "v"])

没有专用函数时用 redis_cmd

px
# 库没封装的命令，直接发
print(redis_cmd(conn, ["SET", "cfg:mode", "prod"]))
print(redis_cmd(conn, ["CONFIG", "GET", "maxmemory"]))
print(redis_cmd(conn, ["EVAL", "return 1", "0"]))

回复按 RESP 类型自动映射：+ 简单串 → str，- 错误 → Err，: 整数 → int，$ 块 → str（$-1 → null），* 数组 → list（*-1 → null）。

坑与边界

1. 一条命令一次往返，没有 pipeline

写 100 个键不要循环 redis_set，用 redis_mset 一次发完；读 100 个用 redis_mget。这是网络往返次数问题，不是客户端优化能救的。

2. 连接不是线程安全的

conn 是带 fd 的 dict。多协程共用一个连接会串包（RESP 是有状态流）。要么每协程一条连接，要么在 Ma 里一次请求一连接。

3. 没有 TLS / RESP3 / 集群 / 哨兵

0.2.0 是 RESP2 明文 TCP 子集。云上要求 rediss:// 的场景用不了；集群键分布要自己算槽。

4. 订阅要长连接

redis_subscribe 之后该连接就进入推送模式，不能再发普通命令，要另开一条连接做写入。

附：40 行写一个 RESP 服务器验证客户端

没有 redis 可连时，用 tcp_listen / tcp_accept 起个假服务器就能端到端验证客户端协议。这也是我实测上面那段输出的办法：

px
def rdline(fd):
    var buf = bytes("")
    while true:
        var r = tcp_recv_ex(fd, 1)
        if not r["ok"] or r["eof"]:
            return null
        buf = bytes_concat(buf, r["data"])
        var n = bytes_len(buf)
        if n >= 2 and bytes_get(buf, n - 2) == 13 and bytes_get(buf, n - 1) == 10:
            return bytes_to_str(bytes_slice(buf, 0, n - 2))

def read_cmd(fd):
    var head = rdline(fd)          # *3
    if head == null or head[0:1] != "*":
        return null
    var n = int(head[1:len(head)])
    var args = []
    while len(args) < n:
        var ln = rdline(fd)        # $4
        var sz = int(ln[1:len(ln)])
        var body = bytes("")
        while bytes_len(body) < sz:
            body = bytes_concat(body, tcp_recv_ex(fd, sz - bytes_len(body))["data"])
        rdline(fd)                 # 尾部 CRLF
        args.append(bytes_to_str(body))
    return args

def main():
    var lfd = tcp_listen(16379)
    while true:
        var cfd = tcp_accept(lfd)
        var store = {}
        while true:
            var a = read_cmd(cfd)
            if a == null:
                break
            var cmd = to_upper(a[0])
            if cmd == "PING":
                tcp_send_ex(cfd, bytes("+PONG\r\n"))
            elif cmd == "GET":
                var v = store.get(a[1], null)
                if v == null:
                    tcp_send_ex(cfd, bytes("$-1\r\n"))
                else:
                    tcp_send_ex(cfd, bytes("$" + str(len(bytes(str(v)))) + "\r\n" + str(v) + "\r\n"))
            elif cmd == "SET":
                store[a[1]] = a[2]
                tcp_send_ex(cfd, bytes("+OK\r\n"))
这段顺带说明了三件事① tcp_listen(port) / tcp_accept(fd) 是内建；② tcp_recv_ex 返回 {ok, eof, data}；③ 二进制读写必须走 bytes 面，bytes_get 返回 0..255 的整数，跟 "\r" 比较要用 13。