Effect 的 Queue 和 PubSub 有什么区别？

同一个 Queue 挂三个消费者，发布一条消息，只有一个消费者拿到它；同样的事换成 PubSub，三个订阅者都会拿到。很多人把 Queue 当广播用，就是栽在这里。

区别在于每条消息给谁：Queue 是分发，每项只被一个消费者取走；PubSub 是广播，每项被每个已订阅的人看到。 想让多个 fiber 分担任务，用 Queue；想让一个事件同时驱动多个独立的反应，用 PubSub。

两者都能限定缓冲大小，也都会在缓冲满时面临「让谁等、丢什么」的选择。下面按这几个问题展开。这里用的 Effect 是 v4 的 API；fiber 是 Effect 里同时运行的执行单元，见 Effect 并发与资源安全。

Queue 适合什么场景？

Queue 是生产者与消费者之间的通道。每个元素只会被一个消费者取走，所以它天然适合任务分发，比如一个入口接收请求，几个 worker fiber 并行处理。

import { Effect, Fiber, Queue } from "effect"

declare const handle: (job: string) => Effect.Effect<void>

export const program = Effect.gen(function*() {
  const jobs = yield* Queue.bounded<string>(100)

  const worker = Effect.forever(
    Queue.take(jobs).pipe(Effect.flatMap(handle))
  )
  const workers = yield* Effect.forEach(
    [1, 2, 3],
    () => Effect.forkChild(worker)
  )

  yield* Queue.offer(jobs, "job-1")
  yield* Queue.offer(jobs, "job-2")

  yield* Fiber.interruptAll(workers)
})

Queue 该选哪种容量策略？

Queue.bounded(n) 是最该优先选择的形态：队列满时，Queue.offer 会挂起生产者，直到消费者取走一些。这是背压，生产者被拖慢，系统整体负载就有了上限，背压在流里的表现见 Effect Stream：背压、分页、扇出与常驻消费者。

其他几种形态是明确的取舍：

Queue.unbounded：永不阻塞，也永远没有上限；生产者快于消费者时内存会持续增长，只在你能证明量级有限时使用。
Queue.dropping(n)：满了丢新元素。
Queue.sliding(n)：满了丢最旧的元素。

收尾时可以用 Queue.end 表示正常结束，再配合 Stream.fromQueue 把队列接成流；Queue.shutdown 则是直接关掉。

PubSub 怎么用，什么时候漏消息？

PubSub 有一个必须理解的时序性质：它只把消息发给已经订阅的人，不是回放日志。在订阅之前发布的消息，之后才订阅的人永远收不到。所以要先建立订阅，再开放发布。订阅需要 Scope，它随着 Scope 关闭而自动退订，Scope 的用法见 Effect 并发与资源安全。

import { Effect, PubSub } from "effect"

export const program = Effect.scoped(
  Effect.gen(function*() {
    const events = yield* PubSub.bounded<string>(64)

    const audit = yield* PubSub.subscribe(events)
    const notify = yield* PubSub.subscribe(events)

    yield* PubSub.publish(events, "user-created")

    const a = yield* PubSub.take(audit)
    const b = yield* PubSub.take(notify)
    return [a, b] // 两个订阅者都拿到了
  })
)

典型用法是一个业务事件同时驱动审计、通知、缓存失效等多个独立的反应。

订阅者慢了，PubSub 会怎样？

和 Queue 一样，bounded 的 PubSub 在缓冲满了之后会让发布方等待最慢的订阅者，所以一个卡住的订阅者能拖住发布方，进而拖住所有人。

dropping 与 sliding 都不会阻塞发布方，但都要接受丢消息，而且丢法不同：

sliding 丢最旧的，快的订阅者仍然收到全部，慢的订阅者只剩最新的几条。
dropping 丢新来的，而「满」是由最慢的订阅者决定的，所以一个慢订阅者会让所有订阅者一起丢掉新消息。

我在 4.0.5 里把这两种行为跑过。需要「慢者不拖累快者」时，sliding 才是对的选择。

事件不能丢的时候该怎么办？

事件绝对不能丢的场景，不应该只靠内存里的 PubSub，要有持久化的来源，比如数据库里的待发事件表。

另外别忘了一个常见的错用：把 Queue 当广播。多个消费者取同一个 Queue 是在分任务，不是都收到；要都收到，用 PubSub。

本文基于 effect 4.0.5。PubSub 的 dropping 与 sliding 行为在该版本上本地跑过；Queue 的容量策略、Queue.end、Stream.fromQueue 的描述以及两段代码示例，没有在这次改写里逐段重新运行验证，使用前请先在你的版本上跑一遍。