从Spring Batch到Flink：实时流处理技术演进与实践

1. 实时流处理的技术革命从分钟级到毫秒级的跨越十年前处理数据时我们还在用定时任务跑批处理今天下单的商品要等到半夜才能进库存系统。现在打开手机应用每笔支付、每次点击都在瞬间完成计算和反馈。这种变化背后是实时流处理技术从实验室走向生产环境的历程。我经历过从Spring Batch到Flink的完整迁移过程。最初用Spring Batch做日终结算后来做小时级对账直到遇到需要实时风控的金融项目时才发现传统批处理架构的瓶颈。当业务要求从T1变成T0从小时级变成秒级再进化到毫秒级时技术栈的升级就成了生死攸关的问题。2. 批处理与流处理的本质差异2.1 Spring Batch的设计哲学Spring Batch作为经典批处理框架其核心设计围绕有限数据集和离散处理两个概念。它的典型工作模式是从数据库或文件读取一批固定数量的记录在内存中进行转换处理将结果写回存储系统重复上述过程直到处理完所有数据这种模式在ETL、报表生成等场景表现优异因为它可以精确控制资源使用如每次处理1000条记录容易实现断点续跑通过JobRepository记录状态对事务有完整支持每个chunk一个事务但当我们尝试用Spring Batch处理实时订单流时立即遇到了几个致命问题实际案例某电商促销活动时用Spring Batch处理订单的惨痛教训即使配置了每分钟触发一次Job高峰期仍积压超过10万订单内存消耗随着队列增长而飙升最终导致OOM风控规则无法实时生效出现大量薅羊毛行为2.2 Flink的流式思维Flink从设计之初就将无限数据流作为一等公民。它的运行时架构决定了几个关键特性事件时间处理每个事件携带自身的时间戳不受处理延迟影响状态管理内置键值存储可以维护窗口状态或会话状态精确一次语义通过检查点机制保证数据不丢不重在同样的电商场景下Flink的表现平均处理延迟50ms从事件产生到触发动作背压机制自动调节处理速度不会OOM支持动态规则更新风控策略秒级生效3. 核心技术对比为什么Flink能实现降维打击3.1 运行时架构差异Spring Batch的架构可以简化为[Reader] - [Processor] - [Writer] ↑ [JobLauncher]这是一个典型的Master-Worker模式每个步骤需要完整执行后才能开始下一步。Flink的架构则是[Source] - [Operator Chain] - [Sink] ↑ ↑ ↑ [TaskManager] [JobManager] [Checkpoint]数据像水流一样持续流动多个操作可以链式合并减少序列化开销。3.2 性能关键指标实测我们在相同硬件环境下对比了两个框架指标Spring BatchFlink吞吐量(events/s)5,000500,00099%延迟(ms)1,20015故障恢复时间(s)603状态大小限制内存限制TB级3.3 典型场景适配性适合Spring Batch的场景银行日终批量清算月度财务报表生成历史数据迁移必须使用Flink的场景实时欺诈检测支付后500ms内判断IoT设备状态监控毫秒级响应实时推荐系统用户浏览时即时计算4. Flink实现毫秒级处理的关键技术4.1 时间语义与窗口机制Flink支持三种时间语义处理时间机器处理事件的系统时间事件时间数据产生时记录的时间戳注入时间数据进入Flink的时间典型的滚动窗口代码示例DataStreamTransaction transactions ... transactions .keyBy(Transaction::getAccountId) .window(TumblingEventTimeWindows.of(Time.seconds(5))) .process(new FraudDetector()) .addSink(new AlertSink());4.2 状态管理与容错Flink的状态后端选择直接影响性能MemoryStateBackend开发测试用FsStateBackend生产环境常用RocksDBStateBackend超大规模状态配置示例state.backend: rocksdb state.checkpoints.dir: hdfs://namenode:8020/flink/checkpoints state.savepoints.dir: hdfs://namenode:8020/flink/savepoints4.3 资源调度优化在K8s环境中部署时这些参数至关重要kubernetes.taskmanager.cpu: 4 taskmanager.numberOfTaskSlots: 4 parallelism.default: 16避坑指南slot数量不是越多越好通常建议设置为CPU核数的70-80%5. 迁移实战从Spring Batch到Flink5.1 思维模式转变批处理思维for (batch in batches) { process(batch); commit(); }流处理思维stream.process(new Function() { void process(event) { // 处理单个事件 } });5.2 代码改造示例Spring Batch版本Bean public ItemProcessorOrder, OrderResult processor() { return order - { RiskEvaluation risk riskService.evaluate(order); return new OrderResult(order, risk); }; }Flink版本DataStreamOrder orders env.addSource(new KafkaSource()); orders.process(new ProcessFunctionOrder, OrderResult() { Override public void processElement(Order order, Context ctx, CollectorOrderResult out) { RiskEvaluation risk riskService.evaluate(order); out.collect(new OrderResult(order, risk)); } });5.3 常见迁移问题解决问题1如何替代Spring Batch的skip逻辑解决方案使用Flink的side output捕获异常数据OutputTagOrder failedOrdersTag new OutputTag(failed-orders); SingleOutputStreamOperatorOrderResult mainStream orders .process(new ProcessFunctionOrder, OrderResult() { Override public void processElement(Order order, Context ctx, CollectorOrderResult out) { try { out.collect(processOrder(order)); } catch (Exception e) { ctx.output(failedOrdersTag, order); } } }); DataStreamOrder failedOrders mainStream.getSideOutput(failedOrdersTag);问题2定时任务如何转换解决方案使用ProcessingTimeTimerpublic class TimerExample extends KeyedProcessFunctionString, Order, Void { Override public void processElement(Order order, Context ctx, CollectorVoid out) { ctx.timerService().registerProcessingTimeTimer(ctx.timestamp() 3600000); } Override public void onTimer(long timestamp, OnTimerContext ctx, CollectorVoid out) { // 每小时执行的操作 } }6. 生产环境调优经验6.1 资源配置黄金法则经过数十个项目的验证我们总结出这些经验值场景TaskManager内存网络缓存并行度低延迟(10ms)4-8GB64MBCPU核数×2高吞吐(100k/s)8-16GB128MBCPU核数×1.5状态密集型16GB32MBCPU核数×0.86.2 检查点配置技巧StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.enableCheckpointing(5000); // 5秒间隔 env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1000); // 最小间隔1秒 env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);6.3 监控指标重点关注通过Prometheus监控这些关键指标numRecordsInPerSecond输入吞吐numRecordsOutPerSecond输出吞吐currentInputWatermark水位线延迟lastCheckpointDuration检查点耗时7. 真实案例秒杀系统优化实录某电商平台在618大促期间将核心系统从Spring Batch迁移到Flink后的变化迁移前峰值QPS2,000平均延迟800ms超时率15%服务器数量20台迁移后峰值QPS50,000平均延迟28ms超时率0.02%服务器数量8台关键优化点使用EventTime处理订单避免时钟不同步问题采用LocalKeyedState实现分布式计数器配置倾斜处理rebalance()rescale()异步IO访问用户风控数据// 异步IO示例 AsyncDataStream.unorderedWait( orders, new AsyncDatabaseRequest(), 1000, // 超时1秒 TimeUnit.MILLISECONDS, 100 // 最大并发请求数 );8. 进阶话题Flink最新特性实践8.1 批流一体新体验Flink 1.16引入的批流统一API// 同样的代码可以跑在流或批模式 ExecutionEnvironment env ExecutionEnvironment.getExecutionEnvironment(); DataStreamString text env.readTextFile(file:///path/to/file); // 或者 StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); DataStreamString text env.addSource(new FileSource(/path/to/file));8.2 CDC连接器实战使用Debezium实现MySQL变更捕获CREATE TABLE products ( id INT, name STRING, price DECIMAL(10,2), PRIMARY KEY (id) NOT ENFORCED ) WITH ( connector mysql-cdc, hostname localhost, port 3306, username flink, password password, database-name inventory, table-name products );8.3 机器学习集成使用Flink ML进行实时预测StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); // 加载模型 DataStreamModel modelStream env.addSource(new ModelSource()); // 数据流 DataStreamFeature featureStream env.addSource(new FeatureSource()); // 实时预测 DataStreamPrediction predictions featureStream .connect(modelStream) .process(new PredictProcessFunction());9. 何时该坚持使用Spring Batch虽然Flink在很多场景下表现优异但Spring Batch仍有其不可替代的优势严格的事务需求需要精细控制每个步骤的事务边界时遗留系统集成已有大量Spring Batch作业且迁移成本过高定时报表生成每天/每周固定时间运行的统计任务简单数据转换不需要复杂状态管理的ETL流程混合架构建议实时链路用Flink处理日终对账用Spring Batch通过消息队列连接两个系统