發(fā)布訂閱、限流與分布式鎖)
后端RPC框架API設計【免費下載鏈接】orpcTypesafe APIs Made Simple 項目地址https://gitcode.com/gh_mirrors/or/orpc點擊查看免費下載導讀orpc/bun是 oRPC 為 Bun 運行時提供的適配器包它不引入任何第三方 Redis 依賴直接基于 Bun 自帶的RedisClient提供三類核心能力基于 Redis Pub/Sub 的發(fā)布訂閱BunRedisPublisher、基于固定窗口計數(shù)器的限流BunRedisRateLimiter以及基于SET NX PX的分布式鎖experimental_BunRedisLocker。讀完本文你將掌握這三個適配器的完整配置參數(shù)、底層實現(xiàn)原理、與 oRPC Procedure/Middleware 的集成方式以及如何通過倉庫內的測試用例驗證其在多進程、多實例環(huán)境下的行為。orpc/bun 在 oRPC 生態(tài)中的定位oRPC 的口號是Typesafe APIs Made Simple整個項目圍繞類型安全的 API展開orpc/contract負責以契約作為單一事實來源orpc/server負責構建 API 或實現(xiàn)契約orpc/client負責端到端類型安全地消費 APIorpc/openapi為 API 增加 OpenAPI 兼容性見 packages/bun/README.md 的 Packages 表格。而orpc/bun屬于Framework ecosystem integrations分組官方定位是Adapters for Buns Redis —— 為 Publisher、Rate Limit、Lock 三個內置能力提供基于 Bun Redis 的適配器。這與 packages/bun/package.json 中包的描述完全一致Bun integration for oRPC: Redis-backed pub/sub, rate limiting, and locking using Buns built-in Redis client也就是說orpc/bun自身不實現(xiàn)業(yè)務邏輯而是把 oRPC 三個通用能力發(fā)布訂閱、限流、鎖的存儲后端替換為 Bun 運行時內置的 Redis 客戶端從而讓 Bun 應用在零額外依賴的前提下獲得跨進程、跨實例共享狀態(tài)的能力。從 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需要滿足兩個前提運行時必須是 Bun適配器直接操作bun模塊導出的RedisClient類型Bun 版本需包含其內置 Redis 客戶端支持需要一個可連接的 Redis 服務測試腳本通過REDIS_URL環(huán)境變量指向 Redis 實例見下文測試與驗證。安裝方式與 oRPC 其他 beta 包一致倉庫內對應命令為參考 apps/content/docs/helpers/publisher.mdx、apps/content/docs/helpers/ratelimit.mdx 與 apps/content/docs/helpers/lock.mdx 的 Installation 小節(jié)npm install orpc/bunbeta由于三個適配器依賴的通用能力分別位于orpc/publisher、orpc/ratelimit、orpc/experimental-lock中實際使用時一般還需要安裝對應的能力包及其 schema 轉換包如orpc/zod。orpc/bun已將其聲明為自身依賴因此引入時會一并可用。BunRedisPublisher基于 Redis Pub/Sub 的跨進程事件分發(fā)BunRedisPublisher是orpc/publisher的 Redis 適配器負責把事件通過 Redis Pub/Sub 分發(fā)到不同進程的訂閱者并可選地借助 Redis Stream 實現(xiàn)錯過的消息可補發(fā)resume?；居梅ê诵膶崿F(xiàn)位于 packages/bun/src/redis-publisher.ts它繼承自 packages/publisher/src/adapters/base-redis.ts 中的抽象基類BaseRedisPublisher泛型參數(shù)T extends Recordstring, 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 }) // 回調式訂閱返回取消訂閱函數(shù) const unsubscribe await publisher.subscribe(something-updated, (payload) { console.log(payload.id) }) await unsubscribe()subscribe同時支持AsyncIterator 風格可直接用于for await...of循環(huán)這在把事件流轉發(fā)給客戶端時非常有用詳見下文與 oRPC Procedure 集成。配置參數(shù)詳解BunRedisPublisher的完整選項由BunRedisPublisherOptionsredis-publisher.ts與BaseRedisPublisherOptionsbase-redis.ts共同定義參數(shù)默認值說明subscriberredis.duplicate()首次訂閱時惰性創(chuàng)建專用于訂閱的 Redis 連接。因為Pub/Sub 會接管連接處于訂閱狀態(tài)的客戶端無法再執(zhí)行普通命令因此必須使用獨立連接prefixRedis Key 與 Pub/Sub 頻道名的前綴用于多應用共享同一 Redis 實例時的隔離serializernew RPCJsonSerializer()負載的序列化/反序列化器默認使用orpc/client的 RPC JSON 序列化器支持自定義 handlers 序列化Date、自定義類等復雜類型resume.enabledfalse是否開啟事件補發(fā)。開啟后發(fā)布的事件會被臨時存入 Redis Stream新訂閱者可基于lastEventId從指定位置恢復resume.seconds3005 分鐘事件保留時長秒。出于性能考慮過期清理是惰性執(zhí)行的所以事件可能比該時長多存活一小段時間此外Publisher基類還提供maxBufferedEvents選項packages/publisher/src/publisher.ts控制 AsyncIterator 訂閱者的緩沖區(qū)上限默認100設為0表示禁用緩沖、事件必須在下一條到達前被消費設為1表示只保留最新事件適合實時狀態(tài)類場景設為Infinity則保留全部事件無丟失但內存占用高。底層原理一條 Lua 腳本保證順序一致BaseRedisPublisher的發(fā)布邏輯base-redis.ts在開啟 resume 時不是先寫 Stream 再 PUBLISH而是通過一條原子 Lua 腳本PUBLISH_SCRIPT完成local idredis.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 寫入順序一致從而避免補發(fā)與實時投遞之間出現(xiàn)順序錯亂。BunRedisPublisher通過redis.send(EVAL, ...)redis-publisher.ts執(zhí)行該腳本并通過XREAD讀取補發(fā)數(shù)據(jù)redis-publisher.ts。在訂閱側base-redis.ts實現(xiàn)順序是先建立 Pub/Sub 訂閱再讀取 Stream 補發(fā)歷史最后處理訂閱期間積壓的實時消息并用resumedIds集合對補發(fā)與實時投遞之間可能競爭的重復事件做去重。這正是 redis-publisher.test.ts 中deduplicates events that race between resume and live delivery during reconnect用例所驗證的行為。BunRedisRateLimiter固定窗口限流BunRedisRateLimiter是orpc/ratelimit的 Redis 適配器實現(xiàn)位于 packages/bun/src/redis-ratelimit.ts繼承自 packages/ratelimit/src/adapters/base-redis.ts 的BaseRedisRateLimiter?；居梅╥mport { 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可據(jù)此拋出 ORPCError(TOO_MANY_REQUESTS, ...) }limit返回的完整結果為{ success, limit, remaining, reset }success表示本次請求是否被允許remaining為窗口內剩余額度下限為 0reset是計數(shù)器重置的時間戳毫秒。配置參數(shù)參數(shù)默認值說明prefixRedis Key 前綴用于隔離不同用途的計數(shù)器maxRequests無必填窗口內允許的最大請求數(shù)window無必填固定窗口時長單位毫秒blockingUntilReady.enabledfalse是否開啟阻塞模式額度不足時等待而不是直接拒絕blockingUntilReady.timeout無enabled 時必填阻塞等待的最大時長毫秒超時后返回success: false底層原理原子 INCRBY 惰性窗口限流核心是一條固定窗口 Lua 腳本FIXED_WINDOW_SCRIPTlocal credis.call(INCRBY,KEYS[1],ARGV[1]) if ctonumber(ARGV[1]) then redis.call(PEXPIRE,KEYS[1],ARGV[2]) end return {c,redis.call(PTTL,KEYS[1])}即對計數(shù)器INCRBY增加本次請求的權重只有當計數(shù)器是新建的c weight才設置PEXPIRE從而以惰性方式啟動窗口最后返回[已用額度, 剩余 TTL]?；惖腸heckLimitbase-redis.ts據(jù)此計算reset Date.now() ttl。值得注意的兩個行為細節(jié)均有測試覆蓋見 redis-ratelimit.test.ts權重校驗limit的weight必須是大于 0 的整數(shù)否則拋出TypeError(Rate limit weight must be an integer greater than 0)base-redis.ts阻塞模式blockUntilReadybase-redis.ts會循環(huán)調用checkLimit若失敗則sleep到reset時刻再試直到成功或超過timeout測試驗證了它在窗口翻轉后放行加權請求、以及reset超出timeout時返回拒絕success: false兩種路徑。experimental_BunRedisLocker基于 SET NX PX 的分布式鎖experimental_BunRedisLocker是orpc/experimental-lock的 Redis 適配器類名帶experimental_前綴說明該能力仍處于實驗階段。實現(xiàn)位于 packages/bun/src/redis-lock.ts繼承自 packages/lock/src/adapters/base-redis.ts 的BaseRedisLocker?；居梅╥mport { 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)在持有鎖期間執(zhí)行回調回調拋出異常時也會釋放鎖finally保證見 base-redis.ts回調參數(shù)waited用于告知是否發(fā)生過等待——如果等待過說明同 key 的其他任務可能剛完成可先查緩存再重算。配置參數(shù)參數(shù)默認值說明prefixRedis Key 前綴ttl無必填鎖的自動過期時間毫秒防止持有者崩潰后鎖永不釋放可在每次調用時覆蓋timeout10000等待鎖可用的最長時間毫秒超時拋出LockTimeoutError可在調用時覆蓋retryInterval100鎖被他人持有時兩次獲取嘗試之間的間隔毫秒signal無可選 AbortSignal中止時提前結束等待并拋出中止原因僅在獲取鎖之前生效底層原理SET NX PX 加鎖 Lua 原子釋放加鎖使用 Redis 原生的SET key token NX PX ttlredis-lock.tsNX保證僅當 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鎖的獲取循環(huán)base-redis.ts在未獲取成功時按retryInterval間隔重試直到timeout到期拋出LockTimeoutError(key)。與 oRPC Procedure / Middleware 的集成這三個適配器與 oRPC 的集成點在apps/content的 helpers 文檔中有完整示例。發(fā)布訂閱在 handler 中直接轉發(fā)事件流參考 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 自動管理——發(fā)布時傳入的事件 id 會被忽略以 Redis Stream 分配的 id 為準而服務端在 yield 自定義負載給客戶端時必須用getEventMeta(payload)?.id取出并隨withEventMeta透傳客戶端重連時才能正確傳回lastEventId續(xù)傳文檔以警告框形式強調了這一點。限流ratelimit 中間件參考 apps/content/docs/helpers/ratelimit.mdxratelimit中間件可基于 context 動態(tài)選擇 limiterimport { 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組合只會執(zhí)行一次限流檢查默認去重配合RateLimitHandlerPlugin還能自動在 HTTP 響應中加入RateLimit-*與Retry-After響應頭。鎖lock 中間件參考 apps/content/docs/helpers/lock.mdxlock中間件讓共享同一 key 的 Procedure 調用互斥執(zhí)行超時未獲取到鎖時 Procedure 以CONFLICT錯誤拒絕請求的signal會被轉發(fā)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互相投遞實時事件、并能從對方發(fā)布的 Stream 中按lastEventId續(xù)傳packages/bun/tests/ratelimit-redis-adapters-compatibility.test.ts驗證二者共享限流計數(shù)器交替調用limit時剩余額度正確遞減、超限后success: falsepackages/bun/tests/lock-redis-adapters-compatibility.test.ts驗證二者共享鎖狀態(tài)——一方持鎖時另一方等待釋放后等待者拿到waited: true。之所以能互通從源碼結構看是因為三個 Bun 適配器都只實現(xiàn)了極薄的協(xié)議層真正復雜的 Key 命名規(guī)則、消息格式、Lua 腳本與重試/補發(fā)邏輯全部沉淀在各自的Base*抽象基類中適配器只負責把 BunRedisClient的命令調用映射為基類需要的四個原語publish、subscribe/unsubscribe、evalScript、readStreamEntries。測試與驗證倉庫為orpc/bun提供了完整的集成測試運行方式在 package.json 中定義bun --env-file../../.env test測試依賴真實 Redis 服務通過REDIS_URL環(huán)境變量指定未設置時測試自動跳過見各測試文件頂部的describe.skipIf(!REDIS_URL)注釋。測試分為三類單元/集成測試redis-publisher.test.ts、redis-ratelimit.test.ts、redis-lock.test.ts覆蓋事件補發(fā)順序、重連去重、并發(fā)發(fā)布者下的 Stream 順序、歷史裁剪與 TTL 過期、加權限流、阻塞模式、鎖超時LockTimeoutError、TTL 到期交接、AbortSignal 中止、以及同一 key 的回調絕不并發(fā)執(zhí)行測試斷言maxActive 1等關鍵行為跨適配器兼容測試上述tests/目錄下的三份*-redis-adapters-compatibility.test.tsRPC 傳輸測試packages/bun/tests/rpc/下還有基于 Bun 原生fetch與 WebSocket 的端到端傳輸測試如client-server.bun-fetch.ts、client-server.bun-websocket.ts用于驗證 oRPC 服務在 Bun 運行時下的完整鏈路。小結orpc/bun以極小的代碼面三個適配器把 Bun 內置 Redis 客戶端接入 oRPC 的發(fā)布訂閱、限流與鎖體系底層通過統(tǒng)一的Base*基類與 Redis Lua 腳本保證原子性發(fā)布腳本保證 Pub/Sub 與 Stream 順序一致、限流腳本保證計數(shù)與窗口原子更新、釋放腳本保證只有 token 持有者可解鎖和跨適配器互通與 node-redis 適配器共享狀態(tài)。對于運行在 Bun 上的 oRPC 應用它是實現(xiàn)實時事件推送、接口限流、冪等任務互斥的首選零依賴方案。補充說明本文所述行為均以當前倉庫代碼為準orpc/bun版本2.0.0-beta.43API 可能隨 beta 迭代調整oRPC 官方文檔入口為 README 中引用的 packages/bun/README.md。oRPC 的設計靈感來自 tRPC端到端類型安全 RPC與 ts-rest契約優(yōu)先與 OpenAPI 集成這一點在 README 的 References 一節(jié)有明確致謝。贊分享后端RPC框架API設計【免費下載鏈接】orpcTypesafe APIs Made Simple 項目地址https://gitcode.com/gh_mirrors/or/orpc點擊查看免費下載上一篇自動化治理架構師實戰(zhàn)指南以 n8n 為中心的自動化審計、風險評估與工作流治理下一篇如何用CSDN博客下載器實現(xiàn)技術知識體系化3步解決內容碎片化難題創(chuàng)作聲明：本文部分內容由AI輔助生成（AIGC），僅供參考 oRPC 的 @orpc/bun 包：用 Bun 內置 Redis 客戶端實現(xiàn)發(fā)布訂閱、限流與分布式鎖
后端RPC框架API設計【免費下載鏈接】orpcTypesafe APIs Made Simple 項目地址https://gitcode.com/gh_mirrors/or/orpc點擊查看免費下載導讀orpc/bun是 oRPC 為 Bun 運行時提供的適配器包它不引入任何第三方 Redis 依賴直接基于 Bun 自帶的RedisClient提供三類核心能力基于 Redis Pub/Sub 的發(fā)布訂閱BunRedisPublisher、基于固定窗口計數(shù)器的限流BunRedisRateLimiter以及基于SET NX PX的分布式鎖experimental_BunRedisLocker。讀完本文你將掌握這三個適配器的完整配置參數(shù)、底層實現(xiàn)原理、與 oRPC Procedure/Middleware 的集成方式以及如何通過倉庫內的測試用例驗證其在多進程、多實例環(huán)境下的行為。orpc/bun 在 oRPC 生態(tài)中的定位oRPC 的口號是Typesafe APIs Made Simple整個項目圍繞類型安全的 API展開orpc/contract負責以契約作為單一事實來源orpc/server負責構建 API 或實現(xiàn)契約orpc/client負責端到端類型安全地消費 APIorpc/openapi為 API 增加 OpenAPI 兼容性見 packages/bun/README.md 的 Packages 表格。而orpc/bun屬于Framework ecosystem integrations分組官方定位是Adapters for Buns Redis —— 為 Publisher、Rate Limit、Lock 三個內置能力提供基于 Bun Redis 的適配器。這與 packages/bun/package.json 中包的描述完全一致Bun integration for oRPC: Redis-backed pub/sub, rate limiting, and locking using Buns built-in Redis client也就是說orpc/bun自身不實現(xiàn)業(yè)務邏輯而是把 oRPC 三個通用能力發(fā)布訂閱、限流、鎖的存儲后端替換為 Bun 運行時內置的 Redis 客戶端從而讓 Bun 應用在零額外依賴的前提下獲得跨進程、跨實例共享狀態(tài)的能力。從 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需要滿足兩個前提運行時必須是 Bun適配器直接操作bun模塊導出的RedisClient類型Bun 版本需包含其內置 Redis 客戶端支持需要一個可連接的 Redis 服務測試腳本通過REDIS_URL環(huán)境變量指向 Redis 實例見下文測試與驗證。安裝方式與 oRPC 其他 beta 包一致倉庫內對應命令為參考 apps/content/docs/helpers/publisher.mdx、apps/content/docs/helpers/ratelimit.mdx 與 apps/content/docs/helpers/lock.mdx 的 Installation 小節(jié)npm install orpc/bunbeta由于三個適配器依賴的通用能力分別位于orpc/publisher、orpc/ratelimit、orpc/experimental-lock中實際使用時一般還需要安裝對應的能力包及其 schema 轉換包如orpc/zod。orpc/bun已將其聲明為自身依賴因此引入時會一并可用。BunRedisPublisher基于 Redis Pub/Sub 的跨進程事件分發(fā)BunRedisPublisher是orpc/publisher的 Redis 適配器負責把事件通過 Redis Pub/Sub 分發(fā)到不同進程的訂閱者并可選地借助 Redis Stream 實現(xiàn)錯過的消息可補發(fā)resume?；居梅ê诵膶崿F(xiàn)位于 packages/bun/src/redis-publisher.ts它繼承自 packages/publisher/src/adapters/base-redis.ts 中的抽象基類BaseRedisPublisher泛型參數(shù)T extends Recordstring, 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 }) // 回調式訂閱返回取消訂閱函數(shù) const unsubscribe await publisher.subscribe(something-updated, (payload) { console.log(payload.id) }) await unsubscribe()subscribe同時支持AsyncIterator 風格可直接用于for await...of循環(huán)這在把事件流轉發(fā)給客戶端時非常有用詳見下文與 oRPC Procedure 集成。配置參數(shù)詳解BunRedisPublisher的完整選項由BunRedisPublisherOptionsredis-publisher.ts與BaseRedisPublisherOptionsbase-redis.ts共同定義參數(shù)默認值說明subscriberredis.duplicate()首次訂閱時惰性創(chuàng)建專用于訂閱的 Redis 連接。因為Pub/Sub 會接管連接處于訂閱狀態(tài)的客戶端無法再執(zhí)行普通命令因此必須使用獨立連接prefixRedis Key 與 Pub/Sub 頻道名的前綴用于多應用共享同一 Redis 實例時的隔離serializernew RPCJsonSerializer()負載的序列化/反序列化器默認使用orpc/client的 RPC JSON 序列化器支持自定義 handlers 序列化Date、自定義類等復雜類型resume.enabledfalse是否開啟事件補發(fā)。開啟后發(fā)布的事件會被臨時存入 Redis Stream新訂閱者可基于lastEventId從指定位置恢復resume.seconds3005 分鐘事件保留時長秒。出于性能考慮過期清理是惰性執(zhí)行的所以事件可能比該時長多存活一小段時間此外Publisher基類還提供maxBufferedEvents選項packages/publisher/src/publisher.ts控制 AsyncIterator 訂閱者的緩沖區(qū)上限默認100設為0表示禁用緩沖、事件必須在下一條到達前被消費設為1表示只保留最新事件適合實時狀態(tài)類場景設為Infinity則保留全部事件無丟失但內存占用高。底層原理一條 Lua 腳本保證順序一致BaseRedisPublisher的發(fā)布邏輯base-redis.ts在開啟 resume 時不是先寫 Stream 再 PUBLISH而是通過一條原子 Lua 腳本PUBLISH_SCRIPT完成local idredis.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 寫入順序一致從而避免補發(fā)與實時投遞之間出現(xiàn)順序錯亂。BunRedisPublisher通過redis.send(EVAL, ...)redis-publisher.ts執(zhí)行該腳本并通過XREAD讀取補發(fā)數(shù)據(jù)redis-publisher.ts。在訂閱側base-redis.ts實現(xiàn)順序是先建立 Pub/Sub 訂閱再讀取 Stream 補發(fā)歷史最后處理訂閱期間積壓的實時消息并用resumedIds集合對補發(fā)與實時投遞之間可能競爭的重復事件做去重。這正是 redis-publisher.test.ts 中deduplicates events that race between resume and live delivery during reconnect用例所驗證的行為。BunRedisRateLimiter固定窗口限流BunRedisRateLimiter是orpc/ratelimit的 Redis 適配器實現(xiàn)位于 packages/bun/src/redis-ratelimit.ts繼承自 packages/ratelimit/src/adapters/base-redis.ts 的BaseRedisRateLimiter?；居梅╥mport { 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可據(jù)此拋出 ORPCError(TOO_MANY_REQUESTS, ...) }limit返回的完整結果為{ success, limit, remaining, reset }success表示本次請求是否被允許remaining為窗口內剩余額度下限為 0reset是計數(shù)器重置的時間戳毫秒。配置參數(shù)參數(shù)默認值說明prefixRedis Key 前綴用于隔離不同用途的計數(shù)器maxRequests無必填窗口內允許的最大請求數(shù)window無必填固定窗口時長單位毫秒blockingUntilReady.enabledfalse是否開啟阻塞模式額度不足時等待而不是直接拒絕blockingUntilReady.timeout無enabled 時必填阻塞等待的最大時長毫秒超時后返回success: false底層原理原子 INCRBY 惰性窗口限流核心是一條固定窗口 Lua 腳本FIXED_WINDOW_SCRIPTlocal credis.call(INCRBY,KEYS[1],ARGV[1]) if ctonumber(ARGV[1]) then redis.call(PEXPIRE,KEYS[1],ARGV[2]) end return {c,redis.call(PTTL,KEYS[1])}即對計數(shù)器INCRBY增加本次請求的權重只有當計數(shù)器是新建的c weight才設置PEXPIRE從而以惰性方式啟動窗口最后返回[已用額度, 剩余 TTL]?；惖腸heckLimitbase-redis.ts據(jù)此計算reset Date.now() ttl。值得注意的兩個行為細節(jié)均有測試覆蓋見 redis-ratelimit.test.ts權重校驗limit的weight必須是大于 0 的整數(shù)否則拋出TypeError(Rate limit weight must be an integer greater than 0)base-redis.ts阻塞模式blockUntilReadybase-redis.ts會循環(huán)調用checkLimit若失敗則sleep到reset時刻再試直到成功或超過timeout測試驗證了它在窗口翻轉后放行加權請求、以及reset超出timeout時返回拒絕success: false兩種路徑。experimental_BunRedisLocker基于 SET NX PX 的分布式鎖experimental_BunRedisLocker是orpc/experimental-lock的 Redis 適配器類名帶experimental_前綴說明該能力仍處于實驗階段。實現(xiàn)位于 packages/bun/src/redis-lock.ts繼承自 packages/lock/src/adapters/base-redis.ts 的BaseRedisLocker?；居梅╥mport { 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)在持有鎖期間執(zhí)行回調回調拋出異常時也會釋放鎖finally保證見 base-redis.ts回調參數(shù)waited用于告知是否發(fā)生過等待——如果等待過說明同 key 的其他任務可能剛完成可先查緩存再重算。配置參數(shù)參數(shù)默認值說明prefixRedis Key 前綴ttl無必填鎖的自動過期時間毫秒防止持有者崩潰后鎖永不釋放可在每次調用時覆蓋timeout10000等待鎖可用的最長時間毫秒超時拋出LockTimeoutError可在調用時覆蓋retryInterval100鎖被他人持有時兩次獲取嘗試之間的間隔毫秒signal無可選 AbortSignal中止時提前結束等待并拋出中止原因僅在獲取鎖之前生效底層原理SET NX PX 加鎖 Lua 原子釋放加鎖使用 Redis 原生的SET key token NX PX ttlredis-lock.tsNX保證僅當 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鎖的獲取循環(huán)base-redis.ts在未獲取成功時按retryInterval間隔重試直到timeout到期拋出LockTimeoutError(key)。與 oRPC Procedure / Middleware 的集成這三個適配器與 oRPC 的集成點在apps/content的 helpers 文檔中有完整示例。發(fā)布訂閱在 handler 中直接轉發(fā)事件流參考 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 自動管理——發(fā)布時傳入的事件 id 會被忽略以 Redis Stream 分配的 id 為準而服務端在 yield 自定義負載給客戶端時必須用getEventMeta(payload)?.id取出并隨withEventMeta透傳客戶端重連時才能正確傳回lastEventId續(xù)傳文檔以警告框形式強調了這一點。限流ratelimit 中間件參考 apps/content/docs/helpers/ratelimit.mdxratelimit中間件可基于 context 動態(tài)選擇 limiterimport { 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組合只會執(zhí)行一次限流檢查默認去重配合RateLimitHandlerPlugin還能自動在 HTTP 響應中加入RateLimit-*與Retry-After響應頭。鎖lock 中間件參考 apps/content/docs/helpers/lock.mdxlock中間件讓共享同一 key 的 Procedure 調用互斥執(zhí)行超時未獲取到鎖時 Procedure 以CONFLICT錯誤拒絕請求的signal會被轉發(fā)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互相投遞實時事件、并能從對方發(fā)布的 Stream 中按lastEventId續(xù)傳packages/bun/tests/ratelimit-redis-adapters-compatibility.test.ts驗證二者共享限流計數(shù)器交替調用limit時剩余額度正確遞減、超限后success: falsepackages/bun/tests/lock-redis-adapters-compatibility.test.ts驗證二者共享鎖狀態(tài)——一方持鎖時另一方等待釋放后等待者拿到waited: true。之所以能互通從源碼結構看是因為三個 Bun 適配器都只實現(xiàn)了極薄的協(xié)議層真正復雜的 Key 命名規(guī)則、消息格式、Lua 腳本與重試/補發(fā)邏輯全部沉淀在各自的Base*抽象基類中適配器只負責把 BunRedisClient的命令調用映射為基類需要的四個原語publish、subscribe/unsubscribe、evalScript、readStreamEntries。測試與驗證倉庫為orpc/bun提供了完整的集成測試運行方式在 package.json 中定義bun --env-file../../.env test測試依賴真實 Redis 服務通過REDIS_URL環(huán)境變量指定未設置時測試自動跳過見各測試文件頂部的describe.skipIf(!REDIS_URL)注釋。測試分為三類單元/集成測試redis-publisher.test.ts、redis-ratelimit.test.ts、redis-lock.test.ts覆蓋事件補發(fā)順序、重連去重、并發(fā)發(fā)布者下的 Stream 順序、歷史裁剪與 TTL 過期、加權限流、阻塞模式、鎖超時LockTimeoutError、TTL 到期交接、AbortSignal 中止、以及同一 key 的回調絕不并發(fā)執(zhí)行測試斷言maxActive 1等關鍵行為跨適配器兼容測試上述tests/目錄下的三份*-redis-adapters-compatibility.test.tsRPC 傳輸測試packages/bun/tests/rpc/下還有基于 Bun 原生fetch與 WebSocket 的端到端傳輸測試如client-server.bun-fetch.ts、client-server.bun-websocket.ts用于驗證 oRPC 服務在 Bun 運行時下的完整鏈路。小結orpc/bun以極小的代碼面三個適配器把 Bun 內置 Redis 客戶端接入 oRPC 的發(fā)布訂閱、限流與鎖體系底層通過統(tǒng)一的Base*基類與 Redis Lua 腳本保證原子性發(fā)布腳本保證 Pub/Sub 與 Stream 順序一致、限流腳本保證計數(shù)與窗口原子更新、釋放腳本保證只有 token 持有者可解鎖和跨適配器互通與 node-redis 適配器共享狀態(tài)。對于運行在 Bun 上的 oRPC 應用它是實現(xiàn)實時事件推送、接口限流、冪等任務互斥的首選零依賴方案。補充說明本文所述行為均以當前倉庫代碼為準orpc/bun版本2.0.0-beta.43API 可能隨 beta 迭代調整oRPC 官方文檔入口為 README 中引用的 packages/bun/README.md。oRPC 的設計靈感來自 tRPC端到端類型安全 RPC與 ts-rest契約優(yōu)先與 OpenAPI 集成這一點在 README 的 References 一節(jié)有明確致謝。贊分享后端RPC框架API設計【免費下載鏈接】orpcTypesafe APIs Made Simple 項目地址https://gitcode.com/gh_mirrors/or/orpc點擊查看免費下載上一篇自動化治理架構師實戰(zhàn)指南以 n8n 為中心的自動化審計、風險評估與工作流治理下一篇如何用CSDN博客下載器實現(xiàn)技術知識體系化3步解決內容碎片化難題創(chuàng)作聲明：本文部分內容由AI輔助生成（AIGC），僅供參考