精确一次语义

工作原理

精确一次通常由至少一次投递加一种抑制第二次副作用的手段构成。常见机制有两种。第一是幂等处理：每个事件携带稳定标识符，接收端记录已应用的标识符，重试就变成空操作。第二是事务处理：消费者在同一个原子事务中写入输出并提交输入偏移量，崩溃时要么两者都保留，要么都不保留。

流处理器还会使用检查点。重启时，作业从最后一个检查点恢复状态，并从对应偏移量继续。如果接收端参与检查点或支持事务提交，管道可以跨故障避免重复；如果不支持，处理器可能重放事件，接收端必须去重。

端到端条件

保证必须在每个边界成立：源、算子状态和接收端。常见失败是消费者先写数据库再提交偏移量。如果进程在两步之间死亡，事件会被重放并写入两次；反转顺序则可能丢失事件。只有跨越两者的事务，或以事件 ID 为键的幂等写入，才能堵住这个缺口。

何时使用

对资金转移、库存扣减、计费计数以及任何无法撤销的副作用使用精确一次处理。对于日志、指标或可重算聚合，至少一次加查询时去重可能更便宜。判断依据是重复的代价，而不是保证听起来是否优雅。

权衡与示例

精确一次增加协调成本：事务、唯一约束、去重表和更长的提交路径，可能降低吞吐并增加延迟。它也无法让发送邮件这类外部副作用变得可逆，外部操作需要自己的幂等键。

示例：消费者把 (event_id, amount) 插入账本表，并以 event_id 作为主键。重投的事件违反主键约束后被忽略。这比分布式事务简单，却能为该写入得到相同的可见结果。

代码示例

export class CsvStreamParser {
  constructor({ delimiter = ",", onRow } = {}) {
    if (typeof delimiter !== "string" || delimiter.length !== 1) {
      throw new TypeError("delimiter must be a single character");
    }
    if (typeof onRow !== "function") {
      throw new TypeError("onRow must be a function");
    }
    this.delimiter = delimiter;
    this.onRow = onRow;
    this.field = "";
    this.row = [];
    this.rowStarted = false;
    this.inQuotes = false;
    this.pendingQuote = false;
    this.pendingCarriageReturn = false;
  }

  write(chunk) {
    const text = String(chunk);
    let startIndex = 0;
    if (this.pendingCarriageReturn) {
      this.pendingCarriageReturn = false;
      if (text.startsWith("\n")) startIndex = 1;
    }
    for (let index = startIndex; index < text.length; index += 1) {
      const char = text[index];
      if (this.inQuotes) {
        if (this.pendingQuote) {
          if (char === '"') {
            this.field += '"';
            this.pendingQuote = false;
            continue;
          }
          this.inQuotes = false;
          this.pendingQuote = false;
        } else if (char === '"') {
          this.pendingQuote = true;
          continue;
        } else {
          this.field += char;
          continue;
        }
      }
      if (char === '"') {
        if (this.field.length > 0) {
          throw new Error("quote must start a field");
        }
        this.inQuotes = true;
        this.rowStarted = true;
      } else if (char === this.delimiter) {
        this.row.push(this.field);
        this.field = "";
        this.rowStarted = true;
      } else if (char === "\n" || char === "\r") {
        if (char === "\r") {
          if (text[index + 1] === "\n") index += 1;
          else this.pendingCarriageReturn = true;
        }
        this.row.push(this.field);
        this.field = "";
        const row = this.row;
        this.row = [];
        this.rowStarted = false;
        this.onRow(row);
      } else {
        this.field += char;
        this.rowStarted = true;
      }
    }
    return this;
  }

  end() {
    if (this.inQuotes && !this.pendingQuote) {
      throw new Error("unterminated quoted field");
    }
    this.inQuotes = false;
    this.pendingQuote = false;
    this.pendingCarriageReturn = false;
    if (this.rowStarted || this.field.length > 0 || this.row.length > 0) {
      this.row.push(this.field);
      this.field = "";
      const row = this.row;
      this.row = [];
      this.rowStarted = false;
      this.onRow(row);
    }
  }
}

const rows = [];
const parser = new CsvStreamParser({ onRow: (row) => rows.push(row) });
parser.write('name,note\r\n"Ada","hello, ');
parser.write('world"\r\n"Grace","said ""hi"""');
parser.end();
console.log(JSON.stringify(rows));

const empty = [];
const quotedEmpty = new CsvStreamParser({ onRow: (row) => empty.push(row) });
quotedEmpty.write('""');
quotedEmpty.end();
console.log(JSON.stringify(empty));

const splitRows = [];
const splitCrlf = new CsvStreamParser({ onRow: (row) => splitRows.push(row) });
splitCrlf.write("a,b\r");
splitCrlf.write("\nc,d");
splitCrlf.end();
console.log(JSON.stringify(splitRows));

try {
  const broken = new CsvStreamParser({ onRow: () => {} });
  broken.write('"open');
  broken.end();
} catch (error) {
  console.log(error.message);
}

from collections.abc import Sequence
from typing import TypeVar

T = TypeVar("T")


def majority_candidate(values: Sequence[T]) -> T | None:
    candidate: T | None = None
    balance = 0
    for value in values:
        if balance == 0:
            candidate = value
            balance = 1
        elif value == candidate:
            balance += 1
        else:
            balance -= 1
    return candidate


def majority_element(values: Sequence[T]) -> T | None:
    if not values:
        return None
    candidate = majority_candidate(values)
    if candidate is None:
        return None
    votes = sum(1 for value in values if value == candidate)
    return candidate if votes > len(values) // 2 else None


def elements_over_third(values: Sequence[T]) -> list[T]:
    if not values:
        return []
    first: T | None = None
    second: T | None = None
    count_first = 0
    count_second = 0
    for value in values:
        if first is not None and value == first:
            count_first += 1
        elif second is not None and value == second:
            count_second += 1
        elif count_first == 0:
            first, count_first = value, 1
        elif count_second == 0:
            second, count_second = value, 1
        else:
            count_first -= 1
            count_second -= 1
    limit = len(values) // 3
    return [
        item
        for item in (first, second)
        if item is not None and values.count(item) > limit
    ]


def main() -> None:
    ballots = ["ada", "bob", "ada", "ada", "cy"]
    print("winner:", majority_element(ballots))
    print("no majority:", majority_element(["a", "b"]))
    print("over a third:", elements_over_third(["x", "x", "y", "y", "z", "x"]))

    records = [
        {"team": "red"},
        {"team": "blue"},
        {"team": "red"},
        {"team": "red"},
    ]
    print("red votes:", majority_element([record["team"] for record in records]))


if __name__ == "__main__":
    main()