调度组合器

调度定义有状态的、可能有副作用的重复事件调度，并以多种方式组合。组合器允许我们获取调度并将它们组合在一起以获得其他调度。

为了演示不同调度的功能，我们将使用以下辅助函数， 它记录每次重复以及相应的延迟（以毫秒为单位），格式为：

#<重复次数>: <延迟毫秒数>

辅助函数（记录执行延迟）

import { Array, Chunk, Duration, Effect, Schedule } from "effect"

const log = (
  schedule: Schedule.Schedule<unknown>,
  delay: Duration.DurationInput = 0
): void => {
  const maxRecurs = 10 // 限制执行次数
  const delays = Chunk.toArray(
    Effect.runSync(
      Schedule.run(
        Schedule.delays(Schedule.addDelay(schedule, () => delay)),
        Date.now(),
        Array.range(0, maxRecurs)
      )
    )
  )
  delays.forEach((duration, i) => {
    console.log(
      i === maxRecurs
        ? "..." // 如果有更多执行，表示截断
        : i === delays.length - 1
        ? "(end)" // 标记最后一次执行
        : `#${i + 1}: ${Duration.toMillis(duration)}ms`
    )
  })
}

组合

调度可以以不同方式组合：

模式	描述
联合	组合两个调度，如果任一调度想要继续则重复，使用较短的延迟。
交集	组合两个调度，仅当两个调度都想要继续时才重复，使用较长的延迟。
序列	通过完全运行第一个调度，然后切换到第二个来组合两个调度。

联合

组合两个调度，如果任一调度想要继续则重复，使用较短的延迟。

示例（组合指数和间隔调度）

import { Array, Chunk, Duration, Effect, Schedule } from "effect"

const log = (
  schedule: Schedule.Schedule<unknown>,
  delay: Duration.DurationInput = 0
): void => {
  const maxRecurs = 10
  const delays = Chunk.toArray(
    Effect.runSync(
      Schedule.run(
        Schedule.delays(Schedule.addDelay(schedule, () => delay)),
        Date.now(),
        Array.range(0, maxRecurs)
      )
    )
  )
  delays.forEach((duration, i) => {
    console.log(
      i === maxRecurs
        ? "..."
        : i === delays.length - 1
        ? "(end)"
        : `#${i + 1}: ${Duration.toMillis(duration)}ms`
    )
  })
}

const schedule = Schedule.union(
  Schedule.exponential("100 millis"),
  Schedule.spaced("1 second")
)

log(schedule)
/*
输出:
#1: 100ms  < exponential
#2: 200ms
#3: 400ms
#4: 800ms
#5: 1000ms < spaced
#6: 1000ms
#7: 1000ms
#8: 1000ms
#9: 1000ms
#10: 1000ms
...
*/

Schedule.union 操作符在每一步选择最短的延迟，所以当组合指数调度和间隔调度时，初始重复将遵循指数退避，然后一旦延迟超过该值就稳定为间隔调度。

交集

组合两个调度，仅当两个调度都想要继续时才重复，使用较长的延迟。

示例（用固定重试次数限制指数退避）

import { Array, Chunk, Duration, Effect, Schedule } from "effect"

const log = (
  schedule: Schedule.Schedule<unknown>,
  delay: Duration.DurationInput = 0
): void => {
  const maxRecurs = 10
  const delays = Chunk.toArray(
    Effect.runSync(
      Schedule.run(
        Schedule.delays(Schedule.addDelay(schedule, () => delay)),
        Date.now(),
        Array.range(0, maxRecurs)
      )
    )
  )
  delays.forEach((duration, i) => {
    console.log(
      i === maxRecurs
        ? "..."
        : i === delays.length - 1
        ? "(end)"
        : `#${i + 1}: ${Duration.toMillis(duration)}ms`
    )
  })
}

const schedule = Schedule.intersect(
  Schedule.exponential("10 millis"),
  Schedule.recurs(5)
)

log(schedule)
/*
输出:
#1: 10ms  < exponential
#2: 20ms
#3: 40ms
#4: 80ms
#5: 160ms
(end)     < recurs
*/

Schedule.intersect 操作符强制执行两个调度的约束。在此示例中，调度遵循指数退避，但由于 Schedule.recurs(5) 限制在5次重复后停止。

序列

通过完全运行第一个调度，然后切换到第二个来组合两个调度。

示例（从固定重试切换到周期性执行）

import { Array, Chunk, Duration, Effect, Schedule } from "effect"

const log = (
  schedule: Schedule.Schedule<unknown>,
  delay: Duration.DurationInput = 0
): void => {
  const maxRecurs = 10
  const delays = Chunk.toArray(
    Effect.runSync(
      Schedule.run(
        Schedule.delays(Schedule.addDelay(schedule, () => delay)),
        Date.now(),
        Array.range(0, maxRecurs)
      )
    )
  )
  delays.forEach((duration, i) => {
    console.log(
      i === maxRecurs
        ? "..."
        : i === delays.length - 1
        ? "(end)"
        : `#${i + 1}: ${Duration.toMillis(duration)}ms`
    )
  })
}

const schedule = Schedule.andThen(
  Schedule.recurs(5),
  Schedule.spaced("1 second")
)

log(schedule)
/*
输出:
#1: 0ms    < recurs
#2: 0ms
#3: 0ms
#4: 0ms
#5: 0ms
#6: 1000ms < spaced
#7: 1000ms
#8: 1000ms
#9: 1000ms
#10: 1000ms
...
*/

第一个调度运行直到完成，之后第二个调度接管。在此示例中，副作用最初执行5次且无延迟，然后每1秒继续一次。

为重试延迟添加随机性

Schedule.jittered 组合器通过在指定范围内应用随机延迟来修改调度。

当资源由于过载或争用而停止服务时，重试和退避对我们没有帮助。如果所有失败的API调用都退避到同一时间点，它们会造成另一次过载或争用。抖动为调度的延迟添加一定量的随机性。这有助于我们避免意外同步并意外使服务停机。

研究 表明 Schedule.jittered(0.0, 1.0) 是在重试中引入随机性的有效方法。

示例（抖动指数退避）

import { Array, Chunk, Duration, Effect, Schedule } from "effect"

const log = (
  schedule: Schedule.Schedule<unknown>,
  delay: Duration.DurationInput = 0
): void => {
  const maxRecurs = 10
  const delays = Chunk.toArray(
    Effect.runSync(
      Schedule.run(
        Schedule.delays(Schedule.addDelay(schedule, () => delay)),
        Date.now(),
        Array.range(0, maxRecurs)
      )
    )
  )
  delays.forEach((duration, i) => {
    console.log(
      i === maxRecurs
        ? "..."
        : i === delays.length - 1
        ? "(end)"
        : `#${i + 1}: ${Duration.toMillis(duration)}ms`
    )
  })
}

const schedule = Schedule.jittered(Schedule.exponential("10 millis"))

log(schedule)
/*
输出:
#1: 10.448486ms
#2: 21.134521ms
#3: 47.245117ms
#4: 88.263184ms
#5: 163.651367ms
#6: 335.818848ms
#7: 719.126709ms
#8: 1266.18457ms
#9: 2931.252441ms
#10: 6121.593018ms
...
*/

Schedule.jittered 组合器在范围内为延迟引入随机性。例如，对指数退避应用抖动确保每次重试在稍微不同的时间发生，减少压垮系统的风险。

使用过滤器控制重复

你可以使用 Schedule.whileInput 或 Schedule.whileOutput 根据应用于其输入或输出的条件来限制调度继续的时间。

示例（基于输出停止）

import { Array, Chunk, Duration, Effect, Schedule } from "effect"

const log = (
  schedule: Schedule.Schedule<unknown>,
  delay: Duration.DurationInput = 0
): void => {
  const maxRecurs = 10
  const delays = Chunk.toArray(
    Effect.runSync(
      Schedule.run(
        Schedule.delays(Schedule.addDelay(schedule, () => delay)),
        Date.now(),
        Array.range(0, maxRecurs)
      )
    )
  )
  delays.forEach((duration, i) => {
    console.log(
      i === maxRecurs
        ? "..."
        : i === delays.length - 1
        ? "(end)"
        : `#${i + 1}: ${Duration.toMillis(duration)}ms`
    )
  })
}

const schedule = Schedule.whileOutput(Schedule.recurs(5), (n) => n <= 2)

log(schedule)
/*
输出:
#1: 0ms < recurs
#2: 0ms
#3: 0ms
(end)   < whileOutput
*/

Schedule.whileOutput 根据调度的输出过滤重复。在此示例中，一旦输出超过 2，调度就会停止，即使 Schedule.recurs(5) 允许最多5次重复。

基于输出调整延迟

Schedule.modifyDelay 组合器允许你根据重复次数或其他输出条件动态更改调度的延迟。

示例（在一定重复次数后减少延迟）

import { Array, Chunk, Duration, Effect, Schedule } from "effect"

const log = (
  schedule: Schedule.Schedule<unknown>,
  delay: Duration.DurationInput = 0
): void => {
  const maxRecurs = 10
  const delays = Chunk.toArray(
    Effect.runSync(
      Schedule.run(
        Schedule.delays(Schedule.addDelay(schedule, () => delay)),
        Date.now(),
        Array.range(0, maxRecurs)
      )
    )
  )
  delays.forEach((duration, i) => {
    console.log(
      i === maxRecurs
        ? "..."
        : i === delays.length - 1
        ? "(end)"
        : `#${i + 1}: ${Duration.toMillis(duration)}ms`
    )
  })
}

const schedule = Schedule.modifyDelay(
  Schedule.spaced("1 second"),
  (out, duration) => (out > 2 ? "100 millis" : duration)
)

log(schedule)
/*
输出:
#1: 1000ms
#2: 1000ms
#3: 1000ms
#4: 100ms  < modifyDelay
#5: 100ms
#6: 100ms
#7: 100ms
#8: 100ms
#9: 100ms
#10: 100ms
...
*/

延迟修改在执行期间动态应用。在此示例中，前三次重复遵循原始的 1秒 间隔。之后，延迟降至 100毫秒，使后续重复更频繁地发生。

点击

Schedule.tapInput 和 Schedule.tapOutput 允许你对调度的输入或输出执行额外的有副作用操作，而不修改其行为。

示例（记录调度输出）

import { Array, Chunk, Duration, Effect, Schedule, Console } from "effect"

const log = (
  schedule: Schedule.Schedule<unknown>,
  delay: Duration.DurationInput = 0
): void => {
  const maxRecurs = 10
  const delays = Chunk.toArray(
    Effect.runSync(
      Schedule.run(
        Schedule.delays(Schedule.addDelay(schedule, () => delay)),
        Date.now(),
        Array.range(0, maxRecurs)
      )
    )
  )
  delays.forEach((duration, i) => {
    console.log(
      i === maxRecurs
        ? "..."
        : i === delays.length - 1
        ? "(end)"
        : `#${i + 1}: ${Duration.toMillis(duration)}ms`
    )
  })
}

const schedule = Schedule.tapOutput(Schedule.recurs(2), (n) =>
  Console.log(`Schedule Output: ${n}`)
)

log(schedule)
/*
输出:
Schedule Output: 0
Schedule Output: 1
Schedule Output: 2
#1: 0ms
#2: 0ms
(end)
*/

Schedule.tapOutput 在每次重复之前运行一个副作用，使用调度的当前输出作为输入。这对于日志记录、调试或触发副作用很有用。