弱项补强 + 编码热身：rate limiter、LRU、并发控制、流式 JSON 解析

// 令牌桶：容量 capacity，每秒匀速补 refillPerSec 个。
// 惰性补充——不起定时器，取的时候按「距上次补充过了多久」一次性补上。
// now 由调用方传入（毫秒），测试才能把时间快进，不用真的 sleep。
class TokenBucket {
  constructor(capacity, refillPerSec, nowMs) {
    this.capacity = capacity
    this.refillPerSec = refillPerSec
    this.tokens = capacity // 冷启动是满桶：允许一次 capacity 大小的突发
    this.lastRefillMs = nowMs
  }
 
  tryAcquire(cost, nowMs) {
    const elapsedMs = Math.max(0, nowMs - this.lastRefillMs) // 时钟回拨不倒扣也不白送
    this.tokens = Math.min(this.capacity, this.tokens + (elapsedMs / 1000) * this.refillPerSec)
    this.lastRefillMs = nowMs
    if (this.tokens < cost) return false
    this.tokens -= cost
    return true
  }
}

import time
from dataclasses import dataclass, field
 
 
# 时间单位是秒，用 time.monotonic 的单调时钟——
# 它不受系统改时间影响，比拿 time.time 当基准省心。
@dataclass
class TokenBucket:
    capacity: float
    refill_per_sec: float
    last_refill: float = field(default_factory=time.monotonic)
    tokens: float = field(init=False)
 
    def __post_init__(self) -> None:
        self.tokens = self.capacity  # 冷启动是满桶
 
    def try_acquire(self, cost: float, now: float) -> bool:
        elapsed = max(0.0, now - self.last_refill)
        self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_per_sec)
        self.last_refill = now
        if self.tokens < cost:
            return False
        self.tokens -= cost
        return True

// synchronized 是单机版必须加的：补充和扣减要在同一个临界区里，
// 否则两个线程都读到「还剩 1 个」，双双放行。时间单位是毫秒。
final class TokenBucket {
    private final double capacity;
    private final double refillPerSec;
    private double tokens;
    private long lastRefillMs;
 
    TokenBucket(double capacity, double refillPerSec, long nowMs) {
        this.capacity = capacity;
        this.refillPerSec = refillPerSec;
        this.tokens = capacity; // 冷启动是满桶
        this.lastRefillMs = nowMs;
    }
 
    synchronized boolean tryAcquire(double cost, long nowMs) {
        long elapsedMs = Math.max(0L, nowMs - lastRefillMs);
        tokens = Math.min(capacity, tokens + elapsedMs / 1000.0 * refillPerSec);
        lastRefillMs = nowMs;
        if (tokens < cost) return false;
        tokens -= cost;
        return true;
    }
}

// 时间单位是秒。跨任务共享时把它包成 actor 或加锁——
// 「补充 + 扣减」必须是一个不可分割的动作，这一点四种语言里都一样。
final class TokenBucket {
    private let capacity: Double
    private let refillPerSec: Double
    private var tokens: Double
    private var lastRefill: Double
 
    init(capacity: Double, refillPerSec: Double, now: Double) {
        self.capacity = capacity
        self.refillPerSec = refillPerSec
        self.tokens = capacity // 冷启动是满桶
        self.lastRefill = now
    }
 
    func tryAcquire(cost: Double, now: Double) -> Bool {
        let elapsed = max(0, now - lastRefill)
        tokens = min(capacity, tokens + elapsed * refillPerSec)
        lastRefill = now
        guard tokens >= cost else { return false }
        tokens -= cost
        return true
    }
}

// JS 的 Map 保插入顺序，所以「提到最新」= delete 再 set，链表可以不写。
class LRUCache {
  constructor(capacity) {
    if (capacity <= 0) throw new Error('capacity 必须大于 0')
    this.capacity = capacity
    this.map = new Map()
  }
 
  get(key) {
    if (!this.map.has(key)) return undefined
    const value = this.map.get(key)
    this.map.delete(key)
    this.map.set(key, value) // 重新插到尾部 = 最近使用
    return value
  }
 
  put(key, value) {
    if (this.map.has(key)) this.map.delete(key) // 更新已有键同样要刷新顺序
    this.map.set(key, value)
    if (this.map.size > this.capacity) {
      const oldest = this.map.keys().next().value // 头部 = 最久未使用
      this.map.delete(oldest)
    }
  }
}

from collections import OrderedDict
 
 
class LRUCache:
    """Python 有 OrderedDict.move_to_end，比手写链表短一半。
    真写业务缓存直接上 functools.lru_cache；这里手写是因为面试要看淘汰顺序。"""
 
    def __init__(self, capacity: int) -> None:
        if capacity <= 0:
            raise ValueError("capacity 必须大于 0")
        self.capacity = capacity
        self.store: OrderedDict[str, int] = OrderedDict()
 
    def get(self, key: str) -> int | None:
        if key not in self.store:
            return None
        self.store.move_to_end(key)  # 移到尾部 = 最近使用
        return self.store[key]
 
    def put(self, key: str, value: int) -> None:
        if key in self.store:
            self.store.move_to_end(key)  # 更新已有键同样要刷新顺序
        self.store[key] = value
        if len(self.store) > self.capacity:
            self.store.popitem(last=False)  # 弹出头部 = 最久未使用

import java.util.LinkedHashMap;
import java.util.Map;
 
// JDK 把这题内置了：accessOrder 传 true 就是 LRU 顺序，
// removeEldestEntry 每次插入后回调一次，返回 true 就淘汰头部。
final class LRUCache<K, V> extends LinkedHashMap<K, V> {
    private final int capacity;
 
    LRUCache(int capacity) {
        super(16, 0.75f, true); // true = 按访问顺序，get 也会把条目挪到尾部
        if (capacity <= 0) throw new IllegalArgumentException("capacity 必须大于 0");
        this.capacity = capacity;
    }
 
    @Override
    protected boolean removeEldestEntry(Map.Entry<K, V> eldest) {
        return size() > capacity;
    }
}

// Swift 标准库没有「保顺序的字典」，只能哈希表 + 双向链表手写。
// 所以这份反而最接近面试官说「不许用内置有序容器」时想看到的答案。
final class LRUCache {
    private final class Node {
        let key: String
        var value: Int
        var next: Node?
        weak var prev: Node? // prev 必须 weak，否则前后互指形成循环引用、永不释放
        init(key: String, value: Int) {
            self.key = key
            self.value = value
        }
    }
 
    private let capacity: Int
    private var map: [String: Node] = [:]
    private let head = Node(key: "", value: 0) // 哨兵，省掉一半空判断
    private let tail = Node(key: "", value: 0)
 
    init(capacity: Int) {
        precondition(capacity > 0, "capacity 必须大于 0")
        self.capacity = capacity
        head.next = tail
        tail.prev = head
    }
 
    func get(_ key: String) -> Int? {
        guard let node = map[key] else { return nil }
        moveToTail(node) // get 也算一次使用
        return node.value
    }
 
    func put(_ key: String, _ value: Int) {
        if let node = map[key] {
            node.value = value
            moveToTail(node)
            return
        }
        let node = Node(key: key, value: value)
        map[key] = node
        insertBeforeTail(node)
        // 先插入再淘汰，顺序反了会在容量为 1 时把刚放进来的自己淘汰掉
        if map.count > capacity, let oldest = head.next, oldest !== tail {
            unlink(oldest)
            map.removeValue(forKey: oldest.key)
        }
    }
 
    private func unlink(_ node: Node) {
        node.prev?.next = node.next
        node.next?.prev = node.prev
        node.next = nil
        node.prev = nil
    }
 
    private func insertBeforeTail(_ node: Node) {
        let last = tail.prev
        last?.next = node
        node.prev = last
        node.next = tail
        tail.prev = node
    }
 
    private func moveToTail(_ node: Node) {
        unlink(node)
        insertBeforeTail(node)
    }
}

// 限制同时运行的异步任务数为 limit。
async function mapWithLimit(items, limit, worker) {
  const results = new Array(items.length)
  let cursor = 0
 
  // 一个 runner 就是一个并发槽位，槽位数固定为 limit
  async function runner() {
    while (cursor < items.length) {
      const index = cursor++
      try {
        results[index] = { ok: true, value: await worker(items[index]) }
      } catch (error) {
        // 失败也必须把槽位还回去：这里不接住，runner 直接退出，
        // 这个槽位就永久空着，并发度从 limit 悄悄掉下去。
        results[index] = { ok: false, reason: String(error) }
      }
    }
  }
 
  await Promise.all(Array.from({ length: Math.min(limit, items.length) }, runner))
  return results
}

import asyncio
from typing import Awaitable, Callable, TypeVar
 
T = TypeVar("T")
 
 
async def map_with_limit(
    items: list[T], limit: int, worker: Callable[[T], Awaitable[object]]
) -> list[dict[str, object]]:
    sem = asyncio.Semaphore(limit)
 
    async def run(item: T) -> dict[str, object]:
        # async with 离开作用域一定 release，等价于 try/finally，
        # 任务抛异常也漏不掉槽位——这是 Python 版最省心的地方。
        async with sem:
            try:
                return {"ok": True, "value": await worker(item)}
            except Exception as exc:
                return {"ok": False, "reason": str(exc)}
 
    # gather 只负责收结果，控并发的是上面那个信号量，不是 gather
    return list(await asyncio.gather(*(run(item) for item in items)))

// 依赖：java.util.concurrent（JDK 17+）
// acquire 放在提交之前：主线程先被挡住，任务队列就不会无限堆积，天然带背压。
static <T, R> List<String> mapWithLimit(List<T> items, int limit, Function<T, R> worker)
        throws InterruptedException {
    var sem = new Semaphore(limit);
    var pool = Executors.newCachedThreadPool();
    var futures = new ArrayList<CompletableFuture<String>>();
    try {
        for (T item : items) {
            sem.acquire();
            futures.add(CompletableFuture.supplyAsync(() -> {
                try {
                    return "ok:" + worker.apply(item);
                } catch (RuntimeException e) {
                    return "err:" + e.getMessage();
                } finally {
                    sem.release(); // 成功失败都要还，这一行就是本题的题眼
                }
            }, pool));
        }
        return futures.stream().map(CompletableFuture::join).toList();
    } finally {
        pool.shutdown();
    }
}

// withTaskGroup 会把 addTask 进去的任务全部并发跑，本身不限流。
// 控并发要自己开窗口：先塞满 limit 个，之后每收一个结果才补一个。
func mapWithLimit<T: Sendable>(
    _ items: [T],
    limit: Int,
    worker: @escaping @Sendable (T) async throws -> String
) async -> [String] {
    var results = [String](repeating: "", count: items.count)
    await withTaskGroup(of: (Int, String).self) { group in
        var next = 0
        func submit(_ index: Int) {
            let item = items[index]
            group.addTask {
                // 错误必须在任务内部收敛成结果值：让它抛出去，
                // 整个 group 会被取消，剩下的任务连槽位带结果一起没了。
                do { return (index, "ok:" + (try await worker(item))) } catch {
                    return (index, "err:\(error)")
                }
            }
        }
        while next < min(limit, items.count) {
            submit(next)
            next += 1
        }
        for await (index, value) in group {
            results[index] = value
            if next < items.count {
                submit(next)
                next += 1
            }
        }
    }
    return results
}

// (a) SSE 那种「一行一个完整 JSON」：行缓冲 + 逐行 parse。
function createLineParser(onEvent) {
  let buffer = ''
  return function feed(chunk) {
    buffer += chunk
    const lines = buffer.split('\n')
    buffer = lines.pop() ?? '' // 最后一段可能是半行，留到下一轮（D1 那个坑）
    for (const line of lines) {
      const trimmed = line.trim()
      if (!trimmed.startsWith('data:')) continue // 空行、注释行、event 行都跳过
      const payload = trimmed.slice(5).trim()
      if (payload === '[DONE]') return // 它不是 JSON，拿去 parse 必炸
      onEvent(JSON.parse(payload))
    }
  }
}

import json
from typing import Any, Callable
 
 
def make_line_parser(on_event: Callable[[dict[str, Any]], None]) -> Callable[[str], None]:
    buffer = ""
 
    def feed(chunk: str) -> None:
        nonlocal buffer
        buffer += chunk
        # 星号解包一步到位：前面的都是完整行，最后一段是半行，留到下一轮
        *lines, buffer = buffer.split("\n")
        for line in lines:
            line = line.strip()
            if not line.startswith("data:"):
                continue
            payload = line[5:].strip()
            if payload == "[DONE]":
                return
            on_event(json.loads(payload))
 
    return feed

// 依赖：Jackson（com.fasterxml.jackson.databind）
// Java 没有「按分隔符切并保留余数」的现成 API，用 StringBuilder 手写行缓冲。
final class LineParser {
    private final StringBuilder buffer = new StringBuilder();
    private final ObjectMapper mapper = new ObjectMapper();
    private final Consumer<JsonNode> onEvent;
 
    LineParser(Consumer<JsonNode> onEvent) {
        this.onEvent = onEvent;
    }
 
    void feed(String chunk) throws JsonProcessingException {
        buffer.append(chunk);
        int nl;
        while ((nl = buffer.indexOf("\n")) >= 0) {
            String line = buffer.substring(0, nl).strip();
            buffer.delete(0, nl + 1); // 剩下的半行留在 buffer 里，等下一轮
            if (!line.startsWith("data:")) continue;
            String payload = line.substring(5).strip();
            if (payload.equals("[DONE]")) return;
            onEvent.accept(mapper.readTree(payload));
        }
    }
}

// 真实代码里 URLSession 的 bytes(for:).lines 已经替你做完了分包与半行拼接（D1 讲过），
// 这里手写一份是为了在白板上写得出来。Swift 有 Codable，直接解成具体类型。
struct StreamEvent: Decodable {
    let delta: String
}
 
final class LineParser {
    private var buffer = ""
    private let decoder = JSONDecoder()
    private let onEvent: (StreamEvent) -> Void
 
    init(onEvent: @escaping (StreamEvent) -> Void) {
        self.onEvent = onEvent
    }
 
    func feed(_ chunk: String) {
        buffer += chunk
        var parts = buffer.components(separatedBy: "\n")
        buffer = parts.removeLast() // 最后一段可能是半行，留到下一轮
        for line in parts {
            let trimmed = line.trimmingCharacters(in: .whitespaces)
            guard trimmed.hasPrefix("data:") else { continue }
            let payload = trimmed.dropFirst(5).trimmingCharacters(in: .whitespaces)
            guard payload != "[DONE]" else { return }
            guard let data = payload.data(using: .utf8),
                  let event = try? decoder.decode(StreamEvent.self, from: data)
            else { continue }
            onEvent(event)
        }
    }
}