工作原理
批处理作业读取有限数据集，通常是文件分区或数据库快照，处理后提交输出。它可以对全部输入排序、连接和重试，并按固定节奏运行。流作业从日志或队列读取事件，为窗口和连接维护状态，并在记录到达时输出结果。它必须处理乱序事件、重启和背压，同时不能丢失进度。
同一种计算常常可以用两种模型表达。每日收入报表可以扫描所有订单；流式版本则维护每分钟的累计值，并在迟到事件出现时修正。
何时分别使用
当分钟或小时级延迟可以接受、输入已经存储，或完整重算是最简单的正确答案时，使用批处理。当决策依赖新鲜事件时使用流处理：欺诈检查、实时看板、告警或运营响应。许多团队同时运行流路径保证新鲜度，再以批路径作为对账的真相来源。
权衡
批处理更容易推理，处理大量数据更便宜，但结果按设计就是过时的。流处理延迟低，却需要状态管理、水位线、精确一次或幂等接收端、延迟监控以及重放策略。在低流量下，流系统通常更贵，因为资源常驻。
示例与误解
支付团队流式处理授权请求，在几秒内拦截可疑卡片，然后运行夜间批处理重新计算已结算收入并发现差异。两条路径共享业务规则，但不共享执行方式。
常见误解是流处理默认就是精确一次。投递保证取决于源、状态后端和接收端。如果接收端不幂等或没有事务，重复处理会重复计数。批处理在作业重跑而没有原子替换输出时，也有类似风险。
代码示例
from collections.abc import Iterator
from typing import Generic, TypeVar
T = TypeVar("T")
class RingBuffer(Generic[T]):
def __init__(self, capacity: int) -> None:
if capacity < 1:
raise ValueError("capacity must be positive")
self._slots: list[T | None] = [None] * capacity
self._head = 0
self._count = 0
@property
def capacity(self) -> int:
return len(self._slots)
def __len__(self) -> int:
return self._count
def push(self, item: T) -> T | None:
evicted: T | None = None
if self._count == self.capacity:
evicted = self._slots[self._head]
self._head = (self._head + 1) % self.capacity
index = (self._head + self._count - 1) % self.capacity
else:
index = (self._head + self._count) % self.capacity
self._slots[index] = item
if self._count < self.capacity:
self._count += 1
return evicted
def items(self) -> Iterator[T]:
for offset in range(self._count):
value = self._slots[(self._head + offset) % self.capacity]
if value is not None:
yield value
def latest(self, count: int) -> list[T]:
if count < 0:
raise ValueError("count must not be negative")
if count == 0:
return []
return list(self.items())[-count:]
def main() -> None:
buffer: RingBuffer[str] = RingBuffer(3)
for event in ["a", "b", "c"]:
print("push", event, "evicted", buffer.push(event))
print("full:", list(buffer.items()))
print("push d evicted:", buffer.push("d"))
print("after:", list(buffer.items()))
print("latest 2:", buffer.latest(2))
print("length:", len(buffer))
if __name__ == "__main__":
main()package main
import (
"fmt"
"math/bits"
)
type Bitset []uint64
func NewBitset(size int) Bitset {
if size < 0 {
panic("negative bitset size")
}
return make(Bitset, (size+63)/64)
}
func (b Bitset) Set(index int) {
b[index/64] |= uint64(1) << (uint(index) % 64)
}
func (b Bitset) Clear(index int) {
b[index/64] &^= uint64(1) << (uint(index) % 64)
}
func (b Bitset) Has(index int) bool {
return b[index/64]&(uint64(1)<<(uint(index)%64)) != 0
}
func (b Bitset) Union(other Bitset) Bitset {
size := len(b)
if len(other) > size {
size = len(other)
}
out := make(Bitset, size)
for i := range out {
var left, right uint64
if i < len(b) {
left = b[i]
}
if i < len(other) {
right = other[i]
}
out[i] = left | right
}
return out
}
func (b Bitset) Intersect(other Bitset) Bitset {
size := len(b)
if len(other) < size {
size = len(other)
}
out := make(Bitset, size)
for i := range out {
out[i] = b[i] & other[i]
}
return out
}
func (b Bitset) Count() int {
total := 0
for _, word := range b {
total += bits.OnesCount64(word)
}
return total
}
func (b Bitset) Slice() []int {
out := make([]int, 0, b.Count())
for wordIndex, word := range b {
for word != 0 {
bit := bits.TrailingZeros64(word)
out = append(out, wordIndex*64+bit)
word &^= uint64(1) << uint(bit)
}
}
return out
}
func main() {
active := NewBitset(80)
for _, id := range []int{1, 5, 64, 79} {
active.Set(id)
}
blocked := NewBitset(80)
blocked.Set(5)
blocked.Set(9)
fmt.Println("count:", active.Count())
fmt.Println("has 64:", active.Has(64), "has 65:", active.Has(65))
fmt.Println("members:", active.Slice())
fmt.Println("union count:", active.Union(blocked).Count())
fmt.Println("intersect:", active.Intersect(blocked).Slice())
active.Clear(64)
fmt.Println("after clear:", active.Count())
}