批处理指南

批处理是一种强大的技术，用于将多个操作分组并作为一个单元进行处理。与确保每个操作单独处理的队列不同，批处理会收集项目并以可配置的组进行处理，从而提高效率并减少开销。本指南涵盖了 TanStack Pacer 的批处理概念。

批处理概念

批处理会随着时间的推移或达到一定大小时收集项目，然后一次性处理所有项目。这非常适合批量处理比逐个处理更有效的场景。批处理可以通过以下方式触发：

达到最大批次大小
等待最长时间
自定义逻辑（例如，特殊项目或条件）

批处理可视化

text

Batching (processing every 3 items or every 2 seconds)
Timeline: [1 second per tick]
Calls:        ⬇️  ⬇️  ⬇️     ⬇️  ⬇️     ⬇️  ⬇️  ⬇️
Batch:       [ABC]   []      [DE]      []      [FGH]  []
Executed:     ✅             ✅         ✅
             [===============================]
             ^ Items are grouped and processed together

             [Items accumulate]   [Process batch]   [Empty]
                in batch           as group         batch

Batching (processing every 3 items or every 2 seconds)
Timeline: [1 second per tick]
Calls:        ⬇️  ⬇️  ⬇️     ⬇️  ⬇️     ⬇️  ⬇️  ⬇️
Batch:       [ABC]   []      [DE]      []      [FGH]  []
Executed:     ✅             ✅         ✅
             [===============================]
             ^ Items are grouped and processed together

             [Items accumulate]   [Process batch]   [Empty]
                in batch           as group         batch

何时使用批处理

批处理的最佳使用场景是：

批量处理项目更有效率（例如，网络请求、数据库写入）
您想减少昂贵操作的频率
您需要控制处理的速率或大小
您想将活动的高峰期进行防抖处理，合并成更少的操作

何时不使用批处理

批处理可能不适合的情况是：

每个项目都必须单独且立即处理（使用队列）
您只关心最新值（使用防抖）

提示

如果您发现自己进行了可以合并的重复调用，批处理可以帮助您优化性能和资源使用。

TanStack Pacer 中的批处理

TanStack Pacer 通过 Batcher 类和简单的 batch 函数提供批处理功能。两者都允许您收集项目并以可配置的批次进行处理。

使用 batch 的基本用法

batch 函数提供了一种创建批处理函数的方式。

import { batch } from '@tanstack/pacer'

// Create a batcher that processes up to 3 items or every 2 seconds
const processBatch = batch<number>(
  (items) => {
    // Process the batch
    console.log('Processing batch:', items)
  },
  {
    maxSize: 3, // Process when 3 items are collected
    wait: 2000, // Or after 2 seconds, whichever comes first
    onItemsChange: (batcher) => {
      console.log('Current batch:', batcher.peekAllItems())
    }
  }
)

// Add items to be batched
processBatch(1)
processBatch(2)
processBatch(3) // Triggers batch processing
processBatch(4)
// Or wait 2 seconds for the next batch to process

import { batch } from '@tanstack/pacer'

// Create a batcher that processes up to 3 items or every 2 seconds
const processBatch = batch<number>(
  (items) => {
    // Process the batch
    console.log('Processing batch:', items)
  },
  {
    maxSize: 3, // Process when 3 items are collected
    wait: 2000, // Or after 2 seconds, whichever comes first
    onItemsChange: (batcher) => {
      console.log('Current batch:', batcher.peekAllItems())
    }
  }
)

// Add items to be batched
processBatch(1)
processBatch(2)
processBatch(3) // Triggers batch processing
processBatch(4)
// Or wait 2 seconds for the next batch to process

注意： 在使用 React 时，优先使用 useBatchedCallback hook 而不是 batch 函数，以实现与 React 生命周期更好的集成和自动清理。

batch 函数返回一个函数，该函数将项目添加到批次中。批次会根据您的配置自动处理。

使用 Batcher 类的进阶用法

Batcher 类提供了对批处理行为的完全控制。

import { Batcher } from '@tanstack/pacer'

// Create a batcher that processes up to 5 items or every 3 seconds
const batcher = new Batcher<number>(
  (items) => {
    // Process the batch
    console.log('Processing batch:', items)
  },
  {
    maxSize: 5, // Process when 5 items are collected
    wait: 3000, // Or after 3 seconds
    getShouldExecute: (items, batcher) => items.includes(42), // Custom trigger
    onItemsChange: (batcher) => {
      console.log('Current batch:', batcher.peekAllItems())
    }
  }
)

// Add items to the batch
batcher.addItem(1)
batcher.addItem(2)
batcher.addItem(3)
// ...

// Manually process the current batch
batcher.execute()

// Control batching
batcher.stop()  // Pause batching
batcher.start() // Resume batching

import { Batcher } from '@tanstack/pacer'

// Create a batcher that processes up to 5 items or every 3 seconds
const batcher = new Batcher<number>(
  (items) => {
    // Process the batch
    console.log('Processing batch:', items)
  },
  {
    maxSize: 5, // Process when 5 items are collected
    wait: 3000, // Or after 3 seconds
    getShouldExecute: (items, batcher) => items.includes(42), // Custom trigger
    onItemsChange: (batcher) => {
      console.log('Current batch:', batcher.peekAllItems())
    }
  }
)

// Add items to the batch
batcher.addItem(1)
batcher.addItem(2)
batcher.addItem(3)
// ...

// Manually process the current batch
batcher.execute()

// Control batching
batcher.stop()  // Pause batching
batcher.start() // Resume batching

Batcher 选项

Batcher 选项允许您自定义批次的处理方式和时间。

maxSize：每个批次的最大项目数（默认值：Infinity）
wait：处理批次前等待的最长时间（毫秒）（默认值：Infinity）
getShouldExecute：一个自定义函数，用于确定是否应处理批次
onExecute：批次处理后的回调
onItemsChange：添加项目或批次处理后的回调
onIsRunningChange：当 batcher 的运行状态改变时触发的回调
started：batcher 是否立即开始运行（默认值：true）

Batcher 方法

Batcher 类提供了几个用于批次管理的方法。

batcher.addItem(item)           // Add an item to the batch
batcher.execute()               // Manually process the current batch
batcher.stop()                  // Pause batching
batcher.start()                 // Resume batching
batcher.store.state.size        // Get current batch size
batcher.store.state.isEmpty     // Check if batch is empty
batcher.store.state.isRunning   // Check if batcher is running
batcher.peekAllItems()           // Get all items in the current batch
batcher.store.state.executionCount // Number of batches processed
batcher.store.state.totalItemsProcessed // Number of items processed
batcher.setOptions(opts)        // Update batcher options
batcher.flush()                 // Flush pending batch immediately

batcher.addItem(item)           // Add an item to the batch
batcher.execute()               // Manually process the current batch
batcher.stop()                  // Pause batching
batcher.start()                 // Resume batching
batcher.store.state.size        // Get current batch size
batcher.store.state.isEmpty     // Check if batch is empty
batcher.store.state.isRunning   // Check if batcher is running
batcher.peekAllItems()           // Get all items in the current batch
batcher.store.state.executionCount // Number of batches processed
batcher.store.state.totalItemsProcessed // Number of items processed
batcher.setOptions(opts)        // Update batcher options
batcher.flush()                 // Flush pending batch immediately

自定义批处理触发器

您可以使用 getShouldExecute 根据自定义逻辑触发批处理。

const batcher = new Batcher<number>(
  (items) => console.log('Processing batch:', items),
  {
    getShouldExecute: (items) => items.includes(99),
  }
)

batcher.addItem(1)
batcher.addItem(99) // Triggers batch processing immediately

const batcher = new Batcher<number>(
  (items) => console.log('Processing batch:', items),
  {
    getShouldExecute: (items) => items.includes(99),
  }
)

batcher.addItem(1)
batcher.addItem(99) // Triggers batch processing immediately

动态配置

您可以在运行时更新 batcher 选项。

batcher.setOptions({
  maxSize: 10,
  wait: 1000,
})

const options = batcher.getOptions()
console.log(options.maxSize) // 10

batcher.setOptions({
  maxSize: 10,
  wait: 1000,
})

const options = batcher.getOptions()
console.log(options.maxSize) // 10

Batcher 类使用 TanStack Store 进行响应式状态管理，提供对批次状态、执行计数和处理状态的实时访问。所有状态都存储在 TanStack Store 中，可以通过 batcher.store.state 访问。但是，如果您使用的是 React 或 Solid 等框架适配器，则不应从此读取状态。而是应该从 batcher.state 读取状态，并为 useBatcher hook 的第三个参数提供一个选择器回调函数来选择加入状态跟踪，如下所示。

状态选择器（框架适配器）

框架适配器支持一个 selector 参数，允许您指定哪些状态更改将触发重新渲染。通过防止不相关的状态更改导致不必要的重新渲染，这可以优化性能。

默认情况下，batcher.state 为空（{}），因为选择器默认为空。 这是 TanStack Store 中的响应式状态 useStore 存储的位置。您必须提供一个选择器函数来选择加入状态跟踪。

// Default behavior - no reactive state subscriptions
const batcher = useBatcher(processFn, { maxSize: 5, wait: 1000 })
console.log(batcher.state) // {}

// Opt-in to re-render when size changes
const batcher = useBatcher(
  processFn, 
  { maxSize: 5, wait: 1000 },
  (state) => ({ size: state.size })
)
console.log(batcher.state.size) // Reactive value

// Multiple state properties
const batcher = useBatcher(
  processFn,
  { maxSize: 5, wait: 1000 },
  (state) => ({
    size: state.size,
    executionCount: state.executionCount,
    status: state.status
  })
)

// Default behavior - no reactive state subscriptions
const batcher = useBatcher(processFn, { maxSize: 5, wait: 1000 })
console.log(batcher.state) // {}

// Opt-in to re-render when size changes
const batcher = useBatcher(
  processFn, 
  { maxSize: 5, wait: 1000 },
  (state) => ({ size: state.size })
)
console.log(batcher.state.size) // Reactive value

// Multiple state properties
const batcher = useBatcher(
  processFn,
  { maxSize: 5, wait: 1000 },
  (state) => ({
    size: state.size,
    executionCount: state.executionCount,
    status: state.status
  })
)

初始状态

您可以在创建 batcher 时提供初始状态值。这通常用于从持久存储中恢复状态。

// Load initial state from localStorage
const savedState = localStorage.getItem('batcher-state')
const initialState = savedState ? JSON.parse(savedState) : {}

const batcher = new Batcher(processFn, {
  maxSize: 5,
  wait: 1000,
  initialState
})

// Load initial state from localStorage
const savedState = localStorage.getItem('batcher-state')
const initialState = savedState ? JSON.parse(savedState) : {}

const batcher = new Batcher(processFn, {
  maxSize: 5,
  wait: 1000,
  initialState
})

订阅状态更改

Store 是响应式的并支持订阅

const batcher = new Batcher(processFn, { maxSize: 5, wait: 1000 })

// Subscribe to state changes
const unsubscribe = batcher.store.subscribe((state) => {
  // do something with the state like persist it to localStorage
})

// Unsubscribe when done
unsubscribe()

const batcher = new Batcher(processFn, { maxSize: 5, wait: 1000 })

// Subscribe to state changes
const unsubscribe = batcher.store.subscribe((state) => {
  // do something with the state like persist it to localStorage
})

// Unsubscribe when done
unsubscribe()

注意： 使用框架适配器时，这没有必要，因为底层的 useStore hook 已经完成了这项工作。如果需要，您也可以从 TanStack Store 导入和使用 useStore 将 batcher.store.state 变成带有自定义选择器的响应式状态。

可用状态属性

BatcherState 包括：

executionCount: 已完成的批处理执行次数
isEmpty：batcher 是否没有待处理的项目（items 数组为空）
isPending: batcher 是否正在等待超时触发批处理
items: 当前排队等待批量处理的项目数组
size: 当前批处理队列中的项目数
status：当前处理状态（“idle” | “pending”）
totalItemsProcessed：已处理的总项目数（跨所有批次）

刷新待处理的批次

batcher 支持刷新待处理的批次以立即触发处理。

const batcher = new Batcher(processFn, { maxSize: 10, wait: 5000 })

batcher.addItem('item1')
batcher.addItem('item2')
console.log(batcher.store.state.isPending) // true

// Flush immediately instead of waiting
batcher.flush()
console.log(batcher.store.state.isEmpty) // true (batch was processed)

const batcher = new Batcher(processFn, { maxSize: 10, wait: 5000 })

batcher.addItem('item1')
batcher.addItem('item2')
console.log(batcher.store.state.isPending) // true

// Flush immediately instead of waiting
batcher.flush()
console.log(batcher.store.state.isEmpty) // true (batch was processed)

框架适配器

每个框架适配器都在 batcher 类之上构建了便捷的 hook 和函数。诸如 useBatcher 或 createBatcher 等 hook 是小的包装器，可以减少您为某些常见用例编写代码所需的样板代码。

有关异步批处理，请参阅异步批处理指南。