—— coding: utf-8 ——

"““流式 vs 批式：延迟与吞吐对比模拟””" import time def run_stream(items, interval=0.001, unit_cost=0.001): "““流式模式：数据到达即处理，逐条输出””“ first_out = None passed = 0 t0 = time.time() for item in items: time.sleep(interval) # 模拟数据到达间隔 time.sleep(unit_cost) # 模拟单条处理开销 if first_out is None: first_out = time.time() - t0 passed += 1 return first_out, time.time() - t0, passed def run_batch(items, batch_size=10, interval=0.001, batch_cost=0.002): ”““批式模式：攒满一批再并行处理””“ first_out = None passed = 0 buffer = [] t0 = time.time() for item in items: time.sleep(interval) # 模拟数据到达间隔 buffer.append(item) if len(buffer) >= batch_size: time.sleep(batch_cost) # 模拟批量并行处理 if first_out is None: first_out = time.time() - t0 passed += len(buffer) buffer = [] return first_out, time.time() - t0, passed def main(): n = 200 items = list(range(n)) fs, ts, sc = run_stream(items) fb, tb, bc = run_batch(items) print(“模式 首结果延迟 总耗时 处理条数”) print(“流式 {:.3f}s {:.3f}s {}条”.format(fs, ts, sc)) print(“批式 {:.3f}s {:.3f}s {}条”.format(fb, tb, bc)) print(“结论：流式首结果快 {:.1f} 倍，批式总吞吐高 {:.1f} 倍”.format( fb / fs, ts / tb)) if name == ”main": main() ```