oRPC 的 @orpc/bun 包：用 Bun 内置 Redis 客户端实现发布订阅、限流与分布式锁

• 后端
• RPC框架
• API设计

导读：@orpc/bun是 oRPC 为 Bun 运行时提供的适配器包，它不引入任何第三方 Redis 依赖，直接基于 Bun 自带的RedisClient提供三类核心能力：基于 Redis Pub/Sub 的发布订阅（BunRedisPublisher）、基于固定窗口计数器的限流（BunRedisRateLimiter）以及基于SET NX PX的分布式锁（experimental_BunRedisLocker）。读完本文，你将掌握这三个适配器的完整配置参数、底层实现原理、与 oRPC Procedure/Middleware 的集成方式，以及如何通过仓库内的测试用例验证其在多进程、多实例环境下的行为。

@orpc/bun 在 oRPC 生态中的定位

oRPC 的口号是Typesafe APIs Made Simple，整个项目围绕"类型安全的 API"展开：@orpc/contract负责以契约作为单一事实来源，@orpc/server负责构建 API 或实现契约，@orpc/client负责端到端类型安全地消费 API，@orpc/openapi为 API 增加 OpenAPI 兼容性（见 packages/bun/README.md 的 Packages 表格）。

而@orpc/bun属于Framework & ecosystem integrations分组，官方定位是：

Adapters for Bun's Redis —— 为 Publisher、Rate Limit、Lock 三个内置能力提供基于 Bun Redis 的适配器。

这与 packages/bun/package.json 中包的描述完全一致：

Bun integration for oRPC: Redis-backed pub/sub, rate limiting, and locking using Bun's built-in Redis client

也就是说，@orpc/bun自身不实现业务逻辑，而是把 oRPC 三个通用能力（发布订阅、限流、锁）的存储后端替换为 Bun 运行时内置的 Redis 客户端，从而让 Bun 应用在"零额外依赖"的前提下获得跨进程、跨实例共享状态的能力。

从 packages/bun/src/index.ts 可以看到，包的公共 API 只有三个导出：

export * from './redis-lock' export * from './redis-publisher' export * from './redis-ratelimit'

同时在 package.json 中，它的运行时依赖仅为@orpc/client、@orpc/experimental-lock、@orpc/publisher、@orpc/ratelimit、@orpc/server、@orpc/shared与@standard-server/core——没有任何 Redis 客户端依赖，因为 Redis 客户端本身由 Bun 运行时提供。

前置条件与安装

使用@orpc/bun需要满足两个前提：

1. 运行时必须是 Bun：适配器直接操作bun模块导出的RedisClient类型，Bun 版本需包含其内置 Redis 客户端支持；
2. 需要一个可连接的 Redis 服务：测试脚本通过REDIS_URL环境变量指向 Redis 实例（见下文"测试与验证"）。

安装方式与 oRPC 其他 beta 包一致，仓库内对应命令为（参考 apps/content/docs/helpers/publisher.mdx、apps/content/docs/helpers/ratelimit.mdx 与 apps/content/docs/helpers/lock.mdx 的 Installation 小节）：

npm install @orpc/bun@beta

由于三个适配器依赖的通用能力分别位于@orpc/publisher、@orpc/ratelimit、@orpc/experimental-lock中，实际使用时一般还需要安装对应的能力包及其 schema 转换包（如@orpc/zod）。@orpc/bun已将其声明为自身依赖，因此引入时会一并可用。

BunRedisPublisher：基于 Redis Pub/Sub 的跨进程事件分发

BunRedisPublisher是@orpc/publisher的 Redis 适配器，负责把事件通过 Redis Pub/Sub 分发到不同进程的订阅者，并可选地借助 Redis Stream 实现"错过的消息可补发"（resume）。

基本用法

核心实现位于 packages/bun/src/redis-publisher.ts，它继承自 packages/publisher/src/adapters/base-redis.ts 中的抽象基类BaseRedisPublisher，泛型参数T extends Record<string, object>用于描述事件名到事件负载的类型映射：

import { BunRedisPublisher } from '@orpc/bun' import { redis } from 'bun' const publisher = new BunRedisPublisher<{ 'something-updated': { id: string } }>(redis, { prefix: 'app:', resume: { enabled: true, seconds: 300, }, }) await publisher.publish('something-updated', { id: '123' }) // 回调式订阅，返回取消订阅函数 const unsubscribe = await publisher.subscribe('something-updated', (payload) => { console.log(payload.id) }) await unsubscribe()

subscribe同时支持AsyncIterator 风格，可直接用于for await...of循环，这在把事件流转发给客户端时非常有用（详见下文"与 oRPC Procedure 集成"）。

配置参数详解

BunRedisPublisher的完整选项由BunRedisPublisherOptions（redis-publisher.ts）与BaseRedisPublisherOptions（base-redis.ts）共同定义：

此外，Publisher基类还提供maxBufferedEvents选项（packages/publisher/src/publisher.ts），控制 AsyncIterator 订阅者的缓冲区上限，默认100：设为0表示禁用缓冲、事件必须在下一条到达前被消费；设为1表示只保留最新事件（适合实时状态类场景）；设为Infinity则保留全部事件（无丢失但内存占用高）。

底层原理：一条 Lua 脚本保证"顺序一致"

BaseRedisPublisher的发布逻辑（base-redis.ts）在开启 resume 时，不是先写 Stream 再 PUBLISH，而是通过一条原子 Lua 脚本（PUBLISH_SCRIPT）完成：

local id=redis.call('XADD',KEYS[1],'*','data',ARGV[1]) if ARGV[2] then redis.call('XTRIM',KEYS[1],'MINID',ARGV[2],ARGV[3]) redis.call('EXPIRE',KEYS[1],ARGV[4]) end redis.call('PUBLISH',KEYS[1],'{"data":'..ARGV[1]..',"id":"'..id..'"}')

即：XADD写入 Stream → 按需XTRIM MINID裁剪过期条目并EXPIRE设置 TTL →PUBLISH到同名频道。脚本的注释明确解释了设计动机：用一条脚本保证 Pub/Sub 投递顺序与 Stream 写入顺序一致，从而避免补发与实时投递之间出现顺序错乱。

BunRedisPublisher通过redis.send('EVAL', ...)（redis-publisher.ts）执行该脚本，并通过XREAD读取补发数据（redis-publisher.ts）。

在订阅侧（base-redis.ts），实现顺序是：先建立 Pub/Sub 订阅，再读取 Stream 补发历史，最后处理订阅期间积压的实时消息，并用resumedIds集合对补发与实时投递之间可能"竞争"的重复事件做去重。这正是 redis-publisher.test.ts 中deduplicates events that race between resume and live delivery during reconnect用例所验证的行为。

BunRedisRateLimiter：固定窗口限流

BunRedisRateLimiter是@orpc/ratelimit的 Redis 适配器，实现位于 packages/bun/src/redis-ratelimit.ts，继承自 packages/ratelimit/src/adapters/base-redis.ts 的BaseRedisRateLimiter。

基本用法

import { BunRedisRateLimiter } from '@orpc/bun' import { redis } from 'bun' const limiter = new BunRedisRateLimiter(redis, { prefix: 'login:', maxRequests: 10, window: 60_000, // 毫秒 }) const result = await limiter.limit('user:123', { weight: 2 }) if (!result.success) { // result 包含 limit / remaining / reset，可据此抛出 ORPCError('TOO_MANY_REQUESTS', ...) }

limit返回的完整结果为{ success, limit, remaining, reset }：success表示本次请求是否被允许，remaining为窗口内剩余额度（下限为 0），reset是计数器重置的时间戳（毫秒）。

配置参数

底层原理：原子 INCRBY + 惰性窗口

限流核心是一条固定窗口 Lua 脚本（FIXED_WINDOW_SCRIPT）：

local c=redis.call('INCRBY',KEYS[1],ARGV[1]) if c==tonumber(ARGV[1]) then redis.call('PEXPIRE',KEYS[1],ARGV[2]) end return {c,redis.call('PTTL',KEYS[1])}

即对计数器INCRBY增加本次请求的权重；只有当计数器是新建的（c == weight）才设置PEXPIRE，从而以惰性方式启动窗口；最后返回[已用额度, 剩余 TTL]。基类的checkLimit（base-redis.ts）据此计算reset = Date.now() + ttl。

值得注意的两个行为细节（均有测试覆盖，见 redis-ratelimit.test.ts）：

• 权重校验：limit的weight必须是大于 0 的整数，否则抛出TypeError('Rate limit weight must be an integer greater than 0')（base-redis.ts）；
• 阻塞模式：blockUntilReady（base-redis.ts）会循环调用checkLimit，若失败则sleep到reset时刻再试，直到成功或超过timeout；测试验证了它在窗口翻转后放行加权请求、以及reset超出timeout时返回拒绝（success: false）两种路径。

experimental_BunRedisLocker：基于 SET NX PX 的分布式锁

experimental_BunRedisLocker是@orpc/experimental-lock的 Redis 适配器，类名带experimental_前缀，说明该能力仍处于实验阶段。实现位于 packages/bun/src/redis-lock.ts，继承自 packages/lock/src/adapters/base-redis.ts 的BaseRedisLocker。

基本用法

import { experimental_BunRedisLocker as BunRedisLocker } from '@orpc/bun' import { redis } from 'bun' const locker = new BunRedisLocker(redis, { ttl: 30_000, timeout: 5_000, }) const report = await locker.lock('report:123', async ({ waited }) => { // waited 为 true 表示曾等待其他持有者释放锁 return await generateReport('123') }, { ttl: 30_000, timeout: 5_000, signal: request.signal, })

lock(key, fn, options)在持有锁期间执行回调，回调抛出异常时也会释放锁（finally保证，见 base-redis.ts），回调参数waited用于告知是否发生过等待——如果等待过，说明同 key 的其他任务可能刚完成，可先查缓存再重算。

配置参数

底层原理：SET NX PX 加锁 + Lua 原子释放

加锁使用 Redis 原生的SET key token NX PX ttl（redis-lock.ts）：NX保证仅当 Key 不存在时才写入（即互斥），PX设置过期时间；返回'OK'表示获取成功。锁的 token 由crypto.randomUUID()生成（base-redis.ts），确保只有持有者本人能释放锁。

释放锁不是简单的DEL，而是通过 Lua 脚本（RELEASE_LOCK_SCRIPT）先比对 token 再删除，避免"持有者 A 的锁已过期、被 B 重新获取后，A 却把 B 的锁删掉"的经典问题：

if redis.call('GET', KEYS[1]) == ARGV[1] then return redis.call('DEL', KEYS[1]) end return 0

锁的获取循环（base-redis.ts）在未获取成功时按retryInterval间隔重试，直到timeout到期抛出LockTimeoutError(key)。

与 oRPC Procedure / Middleware 的集成

这三个适配器与 oRPC 的集成点在apps/content的 helpers 文档中有完整示例。

发布订阅：在 handler 中直接转发事件流

参考 apps/content/docs/helpers/publisher.mdx，事件流可直接作为 Procedure 的输出：

import { os } from '@orpc/server' import * as z from 'zod' const live = os .handler(async function* ({ input, signal, lastEventId }) { const iterator = publisher.subscribe('something-updated', { signal, lastEventId }) for await (const payload of iterator) { yield payload } }) const publish = os .input(z.object({ id: z.string() })) .handler(async ({ input }) => { await publisher.publish('something-updated', { id: input.id }) })

仓库的 playgrounds/bun/src 就是一个完整的 Bun 可运行示例：message.ts中subscribeMessages把publisher.subscribe(channel, { signal, lastEventId })直接作为asyncIteratorObject输出返回，客户端即可实时收到消息。

需要特别注意的是：开启 resume 后，事件 id 由 publisher 自动管理——发布时传入的事件 id 会被忽略（以 Redis Stream 分配的 id 为准）；而服务端在 yield 自定义负载给客户端时，必须用getEventMeta(payload)?.id取出并随withEventMeta透传，客户端重连时才能正确传回lastEventId续传（文档以警告框形式强调了这一点）。

限流：ratelimit 中间件

参考 apps/content/docs/helpers/ratelimit.mdx，ratelimit中间件可基于 context 动态选择 limiter：

import { ratelimit } from '@orpc/ratelimit' const procedure = os .$context<{ ratelimiter: RateLimiter }>() .input(z.object({ email: z.email() })) .use( ratelimit({ limiter: ({ context }) => context.ratelimiter, key: ({ context }, input) => `login:${input.email}`, weight: 1, // 每次请求消耗的额度，默认 1 }), ) .handler(({ input }) => ({ success: true }))

同一请求链中相同limiter + key组合只会执行一次限流检查（默认去重）；配合RateLimitHandlerPlugin还能自动在 HTTP 响应中加入RateLimit-*与Retry-After响应头。

锁：lock 中间件

参考 apps/content/docs/helpers/lock.mdx，lock中间件让共享同一 key 的 Procedure 调用互斥执行，超时未获取到锁时 Procedure 以CONFLICT错误拒绝，请求的signal会被转发：

import { lock } from '@orpc/experimental-lock' const procedure = os .$context<{ locker: Locker }>() .input(z.object({ id: z.string() })) .use( lock({ locker: ({ context }) => context.locker, key: ({ context }, input) => `report:${input.id}`, ttl: 30_000, timeout: 5_000, }), ) .handler(async ({ context, input }) => { if (context['lock/waited']) { // 同 key 的其他调用刚完成，结果可能已可复用 } return await generateReport(input.id) })

跨适配器兼容性：与 Redis 客户端适配器互通

@orpc/bun的三个适配器都不是"独立王国"——它们与@orpc/publisher、@orpc/ratelimit、@orpc/experimental-lock中基于 node-redis 的RedisPublisher/RedisRateLimiter/RedisLocker共享同一套 Redis Key 语义，因此可以在同一集群中混用（例如部分实例跑在 Bun 上、部分跑在 Node 上）。这一点由三份兼容性测试用例背书：

• packages/bun/tests/publisher-redis-adapters-compatibility.test.ts：验证BunRedisPublisher与RedisPublisher互相投递实时事件、并能从对方发布的 Stream 中按lastEventId续传；
• packages/bun/tests/ratelimit-redis-adapters-compatibility.test.ts：验证二者共享限流计数器，交替调用limit时剩余额度正确递减、超限后success: false；
• packages/bun/tests/lock-redis-adapters-compatibility.test.ts：验证二者共享锁状态——一方持锁时另一方等待，释放后等待者拿到waited: true。

之所以能互通，从源码结构看，是因为三个 Bun 适配器都只实现了极薄的协议层：真正复杂的 Key 命名规则、消息格式、Lua 脚本与重试/补发逻辑全部沉淀在各自的Base*抽象基类中，适配器只负责把 BunRedisClient的命令调用映射为基类需要的四个原语（publish、subscribe/unsubscribe、evalScript、readStreamEntries）。

测试与验证

仓库为@orpc/bun提供了完整的集成测试，运行方式在 package.json 中定义：

bun --env-file=../../.env test

测试依赖真实 Redis 服务，通过REDIS_URL环境变量指定（未设置时测试自动跳过，见各测试文件顶部的describe.skipIf(!REDIS_URL)注释）。测试分为三类：

1. 单元/集成测试：redis-publisher.test.ts、redis-ratelimit.test.ts、redis-lock.test.ts，覆盖事件补发顺序、重连去重、并发发布者下的 Stream 顺序、历史裁剪与 TTL 过期、加权限流、阻塞模式、锁超时（LockTimeoutError）、TTL 到期交接、AbortSignal 中止、以及"同一 key 的回调绝不并发执行"（测试断言maxActive === 1）等关键行为；
2. 跨适配器兼容测试：上述tests/目录下的三份*-redis-adapters-compatibility.test.ts；
3. RPC 传输测试：packages/bun/tests/rpc/下还有基于 Bun 原生fetch与 WebSocket 的端到端传输测试（如client-server.bun-fetch.ts、client-server.bun-websocket.ts），用于验证 oRPC 服务在 Bun 运行时下的完整链路。

小结

@orpc/bun以极小的代码面（三个适配器）把 Bun 内置 Redis 客户端接入 oRPC 的发布订阅、限流与锁体系：底层通过统一的Base*基类与 Redis Lua 脚本保证原子性（发布脚本保证 Pub/Sub 与 Stream 顺序一致、限流脚本保证计数与窗口原子更新、释放脚本保证只有 token 持有者可解锁）和跨适配器互通（与 node-redis 适配器共享状态）。对于运行在 Bun 上的 oRPC 应用，它是实现实时事件推送、接口限流、幂等任务互斥的首选零依赖方案。

补充说明：本文所述行为均以当前仓库代码为准（@orpc/bun版本2.0.0-beta.43，API 可能随 beta 迭代调整）；oRPC 官方文档入口为 README 中引用的 packages/bun/README.md。oRPC 的设计灵感来自 tRPC（端到端类型安全 RPC）与 ts-rest（契约优先与 OpenAPI 集成），这一点在 README 的 References 一节有明确致谢。

• 后端
• RPC框架
• API设计