队列

一个 Queue 是一个轻量级的内存队列，具有内置的背压机制，支持异步、纯函数式和类型安全的数据处理。

基本操作

一个 Queue<A> 存储类型为 A 的值，并提供两个基本操作：

API	描述
Queue.offer	向队列添加一个类型为 A 的值。
Queue.take	移除并返回队列中最旧的值。

示例（添加和检索项目）

import { Effect, Queue } from "effect"

const program = Effect.gen(function* () {
  // 创建一个容量为 100 的有界队列
  const queue = yield* Queue.bounded<number>(100)
  // 向队列添加 1
  yield* Queue.offer(queue, 1)
  // 检索并移除最旧的值
  const value = yield* Queue.take(queue)
  return value
})

Effect.runPromise(program).then(console.log)
// 输出: 1

创建队列

队列可以是有界的（具有指定容量）或无界的（没有限制）。不同类型的队列在达到容量时以不同方式处理新值。

有界队列

有界队列在满时应用背压，这意味着任何 Queue.offer 操作都会暂停，直到有空间为止。

示例（创建有界队列）

import { Queue } from "effect"

// 创建一个容量为 100 的有界队列
const boundedQueue = Queue.bounded<number>(100)

丢弃队列

丢弃队列在队列满时丢弃新值。

示例（创建丢弃队列）

import { Queue } from "effect"

// 创建一个容量为 100 的丢弃队列
const droppingQueue = Queue.dropping<number>(100)

滑动队列

滑动队列在达到容量时移除旧值以为新值腾出空间。

示例（创建滑动队列）

import { Queue } from "effect"

// 创建一个容量为 100 的滑动队列
const slidingQueue = Queue.sliding<number>(100)

无界队列

无界队列没有容量限制，允许无限制地添加。

示例（创建无界队列）

import { Queue } from "effect"

// 创建一个没有容量限制的无界队列
const unboundedQueue = Queue.unbounded<number>()

向队列添加项目

offer

使用 Queue.offer 向队列添加值。

示例（添加单个项目）

import { Effect, Queue } from "effect"

const program = Effect.gen(function* () {
  const queue = yield* Queue.bounded<number>(100)
  // 向队列添加 1
  yield* Queue.offer(queue, 1)
})

当使用背压队列时，如果队列满了，Queue.offer 会暂停。为了避免阻塞主纤程，您可以分叉 Queue.offer 操作。

示例（使用 Effect.fork 处理满队列）

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

const program = Effect.gen(function* () {
  const queue = yield* Queue.bounded<number>(1)
  // 用一个项目填充队列
  yield* Queue.offer(queue, 1)
  // 尝试添加第二个项目会暂停，因为队列已满
  const fiber = yield* Effect.fork(Queue.offer(queue, 2))
  // 清空队列以腾出空间
  yield* Queue.take(queue)
  // 加入纤程，完成暂停的 offer
  yield* Fiber.join(fiber)
  // 返回添加后队列的大小
  return yield* Queue.size(queue)
})

Effect.runPromise(program).then(console.log)
// 输出: 1

offerAll

您也可以使用 Queue.offerAll 一次添加多个项目。

示例（添加多个项目）

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

const program = Effect.gen(function* () {
  const queue = yield* Queue.bounded<number>(100)
  const items = Array.range(1, 10)
  // 一次将所有项目添加到队列
  yield* Queue.offerAll(queue, items)
  // 返回添加后队列的大小
  return yield* Queue.size(queue)
})

Effect.runPromise(program).then(console.log)
// 输出: 10

从队列消费项目

take

Queue.take 操作移除并返回队列中最旧的项目。如果队列为空，Queue.take 将暂停，只有在添加项目时才会恢复。为了防止阻塞，您可以将 Queue.take 操作分叉到新纤程中。

示例（在纤程中等待项目）

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

const program = Effect.gen(function* () {
  const queue = yield* Queue.bounded<string>(100)
  // 此 take 操作将暂停，因为队列为空
  const fiber = yield* Effect.fork(Queue.take(queue))
  // 向队列添加一个项目
  yield* Queue.offer(queue, "something")
  // 加入纤程以获取 take 操作的结果
  const value = yield* Fiber.join(fiber)
  return value
})

Effect.runPromise(program).then(console.log)
// 输出: something

poll

要在不暂停的情况下检索队列的第一个项目，请使用 Queue.poll。如果队列为空，Queue.poll 返回 None；如果有项目，它将其包装在 Some 中。

示例（轮询项目）

import { Effect, Queue } from "effect"

const program = Effect.gen(function* () {
  const queue = yield* Queue.bounded<number>(100)
  // 向队列添加项目
  yield* Queue.offer(queue, 10)
  yield* Queue.offer(queue, 20)
  // 如果可用，检索第一个项目
  const head = yield* Queue.poll(queue)
  return head
})

Effect.runPromise(program).then(console.log)
/*
输出:
{
  _id: "Option",
  _tag: "Some",
  value: 10
}
*/

takeUpTo

要检索多个项目，请使用 Queue.takeUpTo，它返回最多指定数量的项目。 如果没有足够的项目，它会返回所有可用项目而不等待更多。

当不需要确切数量的项目时，此函数对批处理特别有用。它确保程序继续处理当前可用的任何数据。

如果您需要在继续之前等待确切数量的项目，请考虑使用 takeN。

示例（取最多 N 个项目）

import { Effect, Queue } from "effect"

const program = Effect.gen(function* () {
  const queue = yield* Queue.bounded<number>(100)

  // 向队列添加项目
  yield* Queue.offer(queue, 1)
  yield* Queue.offer(queue, 2)
  yield* Queue.offer(queue, 3)

  // 检索最多 2 个项目
  const chunk = yield* Queue.takeUpTo(queue, 2)
  console.log(chunk)

  return "some result"
})

Effect.runPromise(program).then(console.log)
/*
输出:
{ _id: 'Chunk', values: [ 1, 2 ] }
some result
*/

takeN

从队列中取指定数量的元素。如果队列没有足够的元素，操作会暂停，直到所需数量的元素可用。

此函数对于需要一次处理确切数量项目的场景很有用，确保操作在批次完成之前不会继续。

示例（取固定数量的项目）

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

const program = Effect.gen(function* () {
  // 创建一个可容纳最多 100 个元素的队列
  const queue = yield* Queue.bounded<number>(100)

  // 分叉一个尝试从队列取 3 个项目的纤程
  const fiber = yield* Effect.fork(
    Effect.gen(function* () {
      console.log("尝试从队列取 3 个项目...")
      const chunk = yield* Queue.takeN(queue, 3)
      console.log(`成功取了 3 个项目: ${chunk}`)
    })
  )

  // 最初只提供 2 个项目
  yield* Queue.offer(queue, 1)
  yield* Queue.offer(queue, 2)
  console.log(
    "提供了 2 个项目。纤程现在正在等待第 3 个项目..."
  )

  // 模拟一些延迟
  yield* Effect.sleep("2 seconds")

  // 提供第 3 个项目，这将解除 takeN 调用的阻塞
  yield* Queue.offer(queue, 3)
  console.log("提供了第 3 个项目，这应该解除纤程的阻塞。")

  // 等待纤程完成
  yield* Fiber.join(fiber)
  return "some result"
})

Effect.runPromise(program).then(console.log)
/*
输出:
提供了 2 个项目。纤程现在正在等待第 3 个项目...
尝试从队列取 3 个项目...
提供了第 3 个项目，这应该解除纤程的阻塞。
成功取了 3 个项目: {
  "_id": "Chunk",
  "values": [
    1,
    2,
    3
  ]
}
some result
*/

takeAll

要一次从队列中检索所有项目，请使用 Queue.takeAll。此操作立即完成，如果队列为空则返回空集合。

示例（取所有项目）

import { Effect, Queue } from "effect"

const program = Effect.gen(function* () {
  const queue = yield* Queue.bounded<number>(100)
  // 向队列添加项目
  yield* Queue.offer(queue, 10)
  yield* Queue.offer(queue, 20)
  yield* Queue.offer(queue, 30)
  // 从队列检索所有项目
  const chunk = yield* Queue.takeAll(queue)
  return chunk
})

Effect.runPromise(program).then(console.log)
/*
输出:
{
  _id: "Chunk",
  values: [ 10, 20, 30 ]
}
*/

关闭队列

shutdown

Queue.shutdown 操作允许您中断当前在 offer* 或 take* 操作上暂停的所有纤程。此操作还会清空队列，并使任何未来的 offer* 和 take* 调用立即终止。

示例（在队列关闭时中断纤程）

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

const program = Effect.gen(function* () {
  const queue = yield* Queue.bounded<number>(3)
  // 分叉一个等待从队列取项目的纤程
  const fiber = yield* Effect.fork(Queue.take(queue))
  // 关闭队列，中断纤程
  yield* Queue.shutdown(queue)
  // 加入被中断的纤程
  yield* Fiber.join(fiber)
})

awaitShutdown

Queue.awaitShutdown 操作可用于在队列关闭时运行 effect。它等待直到队列关闭，如果队列已经关闭则立即恢复。

示例（等待队列关闭）

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

const program = Effect.gen(function* () {
  const queue = yield* Queue.bounded<number>(3)
  // 分叉一个纤程以等待队列关闭并记录消息
  const fiber = yield* Effect.fork(
    Queue.awaitShutdown(queue).pipe(
      Effect.andThen(Console.log("正在关闭"))
    )
  )
  // 关闭队列，触发纤程中的等待
  yield* Queue.shutdown(queue)
  yield* Fiber.join(fiber)
})

Effect.runPromise(program)
// 输出: 正在关闭

仅提供/仅取队列

有时，您可能希望代码的某些部分只向队列添加值（Enqueue）或只从队列检索值（Dequeue）。Effect 提供接口来强制执行这些特定功能。

Enqueue

向队列添加值的所有方法都由 Enqueue 接口定义。这将队列限制为仅提供操作。

示例（将队列限制为仅提供操作）

import { Queue } from "effect"

const send = (offerOnlyQueue: Queue.Enqueue<number>, value: number) => {
  // 此队列仅限于提供操作

  // 错误：不能在仅提供队列上使用 take
  Queue.take(offerOnlyQueue)
Error ts(2345)  ― Argument of type 'Enqueue<number>' is not assignable to parameter of type 'Dequeue<unknown>'.
  Type 'Enqueue<number>' is missing the following properties from type 'Dequeue<unknown>': take, takeAll, takeUpTo, takeBetween, and 6 more.

  // 有效的提供操作
  return Queue.offer(offerOnlyQueue, value)
}

Dequeue

类似地，从队列检索值的所有方法都由 Dequeue 接口定义，该接口将队列限制为仅取操作。

示例（将队列限制为仅取操作）

import { Queue } from "effect"

const receive = (takeOnlyQueue: Queue.Dequeue<number>) => {
  // 此队列仅限于取操作

  // 错误：不能在仅取队列上使用 offer
  Queue.offer(takeOnlyQueue, 1)
Error ts(2345)  ― Argument of type 'Dequeue<number>' is not assignable to parameter of type 'Enqueue<number>'.
  Type 'Dequeue<number>' is missing the following properties from type 'Enqueue<number>': offer, unsafeOffer, offerAll, [EnqueueTypeId]

  // 有效的取操作
  return Queue.take(takeOnlyQueue)
}

The Queue 类型结合了 Enqueue 和 Dequeue，因此您可以轻松地将其传递给代码的不同部分，根据需要仅强制执行 Enqueue 或 Dequeue 行为。

示例（一起使用仅提供和仅取队列）

import { Effect, Queue } from "effect"

const send = (offerOnlyQueue: Queue.Enqueue<number>, value: number) => {
  return Queue.offer(offerOnlyQueue, value)
}

const receive = (takeOnlyQueue: Queue.Dequeue<number>) => {
  return Queue.take(takeOnlyQueue)
}

const program = Effect.gen(function* () {
  const queue = yield* Queue.unbounded<number>()

  // 向队列添加值
  yield* send(queue, 1)
  yield* send(queue, 2)

  // 从队列检索值
  console.log(yield* receive(queue))
  console.log(yield* receive(queue))
})

Effect.runFork(program)
/*
输出:
1
2
*/