做 Flink 实时开发的这几年双流操作是绕不过去的一个坎。不管是订单流和支付流做实时对账还是用户点击流和曝光流做归因分析本质上都是把两条独立的数据流按照某个 key 和时间范围关联起来。我在带团队时也发现很多人会把 DataStream API 用得很熟但一旦被问到 flink 双流操作源码是怎么把两条流“合”起来的就支支吾吾了。这篇文章我从源码层面把双流操作的主线拆一遍适合已经能熟练使用 DataStream API、想深入理解双流 join 内部机制的开发者也适合正在排查双流 join 数据延迟、状态爆炸问题的同学参考。1. 双流操作的入口从 DataStream API 到 ConnectedStreams1.1 双流到底有哪些玩法先说清楚一件事Flink 里的“双流操作”不是只有 join 一个词。表面上大家都说“双流 join”源码里其实分成了四条路线。第一个是connect它把两条流连接成一个ConnectedStreams之后可以调用process传入CoProcessFunction完全自定义两条流的处理逻辑。这是最灵活的方式因为两条流可以各自维护自己的状态也可以互相访问对方的数据。第二个是coGroup按 key 分组后把两条流放在同一个窗口里做分组关联输出的是一个集合。它适合做“左流和右流各自聚合后再拼接”这种场景比如算出每个用户今天的下单笔数和支付笔数然后合到一条结果里。第三个是join基于窗口的等值 join。它底层本质上是coGroup的一种特化窗口内的匹配规则更严格要求左右两条流的数据落在同一个窗口且 key 相等。第四个是intervalJoin也叫区间 join它不走窗口而是给左流记录设置一个上下界时间范围在范围内找右流匹配记录。它对应的底层算子是IntervalJoinOperator这也是我最推荐在源码层面仔细读的一个类。很多初学者会以为这些 API 是各自独立实现的其实connect才是老祖宗。coGroup、join、intervalJoin最终都会被编译成某种“双输入算子”只是内部匹配逻辑不同。理解了connect这条主线其他三条路线就能很快串起来。1.2 源码里 connect 之后发生了什么直接看代码链路。你写这样一段逻辑DataStreamOrder orderStream ...; DataStreamPay payStream ...; orderStream.connect(payStream) .keyBy(Order::getOrderId, Pay::getOrderId) .process(new MyCoProcessFunction())第一步orderStream.connect(payStream)返回一个ConnectedStreams对象。这个对象特别简单没有复杂的计算逻辑就是拿两个DataStream包一层同时保存两个流各自的数据类型TypeInformation。第二步.keyBy(...)依然返回ConnectedStreams只不过内部记录了两个 KeySelector以及 key 对应的 TypeInformation。第三步.process(new MyCoProcessFunction())时ConnectedStreams会调用内部的transform方法public R SingleOutputStreamOperatorR process( CoProcessFunctionIN1, IN2, R function, TypeInformationR outputType) { TwoInputStreamOperatorIN1, IN2, R operator new CoProcessOperator(clean(function)); return transform(Co-Process, outputType, operator); }注意这里transform的最后一个参数类型是TwoInputStreamOperator。这是整个双流操作源码里非常核心的一个接口后面所有双流处理逻辑都是围绕它展开的。transform方法内部会创建新的Transformation节点把两个输入 Transformation 挂上去最终交给 StreamGraphGenerator 生成作业图。也就是说connect本身不产生任何计算真正干活的是TwoInputStreamOperator。只要想明白这个接口的设计双流操作的基础架构就通了。1.3 为什么所有双流操作都要落到 TwoInputStreamOperatorFlink 的运行时模型里单个算子只能处理一个输入流的记录对应OneInputStreamOperator。如果要处理两个输入就必须有一个接口同时暴露两个processElement入口。TwoInputStreamOperator就是这个入口协议。它的核心方法只有两个void processElement1(StreamRecordIN1 element) throws Exception; void processElement2(StreamRecordIN2 element) throws Exception;当第一个输入流来一条数据运行时框架就会调用processElement1第二个输入流的数据进入时调用processElement2。除此之外它还要处理水印、检查点屏障、算子状态快照等生命周期方法。先搞懂这个接口再去看CoProcessOperator、KeyedCoProcessOperator、IntervalJoinOperator你就会发现它们本质上都是在实现这两个方法。区别只是内部缓存、匹配逻辑、状态管理方式不一样。2. 所有双流操作的地基CoProcessFunction 与 KeyedCoProcessOperator2.1 CoProcessFunction 的两个处理方法是两条流的汇合点CoProcessFunction是你在 API 层接触最多的双流处理函数它的抽象方法就两个processElement1和processElement2。每个方法里拿到的Context都能做三件关键的事情第一访问当前 key。即使两条流是通过不同的字段 keyBy 的进入CoProcessFunction后都会被统一成同一个 key 空间你在函数里通过ctx.getCurrentKey()拿到的就是当前数据的 key。第二访问TimerService。你可以注册事件时间定时器或处理时间定时器定时器触发后会回调onTimer方法。这个机制是双流 join 能实现“等待一段时间再匹配”的关键抓手。第三输出到侧输出流。双流处理中经常有“匹配不上”的孤儿数据直接用ctx.output发到侧输出流比在主流里塞脏数据优雅得多。我经常看到有人把CoProcessFunction理解成“两个 onElement 拼在一起”这是不够的。它实际上是两条流的“汇合点”也是一个可以自由控制状态的“独立算子”。2.2 KeyedCoProcessOperator 如何处理两个输入在 keyBy 之后调用process真正被创建出来的算子不是CoProcessOperator而是KeyedCoProcessOperator。它同样实现了TwoInputStreamOperator但多了 keyed state 和 keyed timer 的支持。看KeyedCoProcessOperator.processElement1的核心实现逻辑Override public void processElement1(StreamRecordIN1 element) throws Exception { collector.setTimestamp(element); context.element element; userFunction.processElement1(element.getValue(), context, collector); }代码本身不复杂但有一个特别容易被忽略的点在调用userFunction之前运行时框架已经通过setCurrentKey把当前 key 设置到了算子后端的状态存储里。也就是说你在CoProcessFunction里拿到ValueState去读数据时读到的永远是“当前 key 对应的那个 value”。这个 key 隔离机制是由KeyedCoProcessOperator所在的上下文保证的不是CoProcessFunction自己处理的。processElement2的逻辑完全对称。两个方法的执行是串行的不可能出现同一个算子同时执行processElement1和processElement2的情况。同一时刻只有一条记录在算子内处理这点对理解状态并发安全非常重要。2.3 状态编程在双流操作里的落地方式双流 join 最常见的状态模式是左流来了先存状态右流来了去状态里找匹配匹配成功后可能还要把状态删掉。所以代码里经常出现ValueState存“最新一条记录”或者MapState存“一个 key 对应的多条记录”。我自己在实际项目里踩过一个坑用ValueState存左流记录右流数据量比左流大好几倍状态里永远只保留左流最新一条导致旧记录永远匹配不上。后来换成MapStateLong, Orderkey 用事件时间戳才能在一个 key 下缓存多条记录。源码层面的规律是KeyedCoProcessOperator只是把状态 api 暴露给你它不会主动帮你清理状态。所有状态的过期删除都得自己在onTimer里处理或者用 TTL 配置让状态自动过期。这个问题到了第 5 章我会重点讲。3. Interval Join 源码深度拆解定时器加缓冲区的组合逻辑3.1 Interval Join 的算子长什么样先看 API。双流 interval join 的写法是orderStream .keyBy(Order::getOrderId) .intervalJoin(payStream.keyBy(Pay::getOrderId)) .between(Time.minutes(-5), Time.minutes(5)) .process(new ProcessJoinFunctionOrder, Pay, Result() { ... });这段代码最终会生成一个IntervalJoinOperator。它和KeyedCoProcessOperator最大的区别在于IntervalJoinOperator内部自己管理了两份缓存状态并且注册了很多定时器来做缓冲区的清理。IntervalJoinOperator的核心状态有三个leftBuffer一个ListStateBufferEntryLEFT缓存左流记录rightBuffer一个ListStateBufferEntryRIGHT缓存右流记录timerState记录每个缓冲区条目的定时器时间戳用于清理时核对BufferEntry里存的不只是原始数据还带了一个时间戳。因为 interval join 的时间匹配要基于事件时间所以每条记录进入算子时它的timestamp会被单独存下来后续所有时间判断都靠这个字段。3.2 左右流如何互相寻找匹配记录当一条左流记录leftRecord到达时IntervalJoinOperator.processElement1会做三件事。第一把leftRecord封装成BufferEntry加入leftBuffer状态防止后面右流数据到了之后找不到它。第二遍历当前rightBuffer里的每一条右流记录rightRecord判断时间差是否在[lowerBound, upperBound]区间内。判断逻辑是rightRecord.timestamp - leftRecord.timestamp是否落在between指定的上下界。第三如果时间差满足条件就把左右两条记录交给用户定义的ProcessJoinFunction.join方法如果不满足就继续遍历下一条。右流记录到达时的处理完全对称先缓存自己再遍历左流缓冲区找匹配。这里有一个设计上的关键点interval join 的匹配判断完全基于“先缓存、后遍历”的方式而不是像窗口 join 那样等窗口结束才统一触发。所以它的延迟更低数据一到就可能触发 join。但代价就是需要维护大量的缓冲区条目和定时器。3.3 状态清理的触发时机与过期数据处理如果只往缓冲区里加数据不清理状态迟早会爆。所以IntervalJoinOperator为每条进入缓冲区的记录都注册了一个事件时间定时器。定时器的触发时间是这样设计的对于一条左流记录它的最大可匹配右流记录时间范围是[left.timestamp lowerBound, left.timestamp upperBound]。当算子水位推进到left.timestamp upperBound之后这条左流记录就再也不可能被任何右流记录匹配了此时应该把它从leftBuffer中移除。具体实现时算子会注册一个定时器触发时间就是这个“上限时间”。定时器回调里会把对应的BufferEntry从缓冲区状态里删掉。这个机制让我觉得特别值得学它不是事后扫描全部状态来清理而是提前算好每条记录的“死亡时间”到点精准删除。理解了这一点你在自己写CoProcessFunction做双流缓存时也应该采用同样的模式每条数据进来就注册清理定时器而不是等状态变大后再做全量清理。还有一个小细节容易被坑定时器的触发依赖事件时间水位。如果某条流的事件时间一直不推进比如lowerBound是负数而用户数据时间戳远大于当前水位清理定时器会迟迟不触发缓冲区会一直膨胀。所以 interval join 上线前必须检查两条流的事件时间字段是否合理、是否有乱序数据导致水位长时间不动。3.4 一个参数范围误导引发的血案我印象很深的一次事故是某团队在写 interval join 时把between参数理解反了。between(-5, 5)表示左流记录可以和“时间戳相差 5 分钟以内”的右流记录关联。但问题的关键是参照系是左流记录。业务方期望的是“支付发生在下单后 10 分钟内”于是代码写了between(-10, 0)以为 0 到 10 分钟。结果运行后发现大量支付记录匹配不上。原因是左流是支付流时between(-10, 0)的意思是支付时间戳减去下单时间戳的范围在[-10, 0]分钟也就是“支付不能晚于下单”这明显是反的。正确做法应该是先确定哪条流是时间基准流然后按“另一条流时间戳 − 基准流时间戳”的差值写范围。如果拿支付流当基准期望支付在下单后 0 到 10 分钟内就要写between(0, 10)。这个参数问题在源码里多看两行就能避免因为IntervalJoinOperator内部比较顺序非常清晰。4. 双流操作中的时间与容错水印对齐与检查点4.1 两个输入的水印为什么要对齐双流算子有两个输入每个输入源都有自己的水位。如果算子直接采用任意一条流的水位来处理定时器会出现严重的时间错乱。比如左流水位已经到 12:00右流水位才到 11:30。此时左流注册了一个定时器在 11:50如果直接拿左流水位触发这个定时器就会立刻触发可右流 11:50 的数据可能还没到匹配就会不完整。Flink 的办法是多输入算子会为每个输入分别维护当前水位然后取最小值作为算子当前的事件时间水位。只有两条流的水位都推进到某个时间点定时器才会在这个时间点触发。源码里这块逻辑在TwoInputStreamProcessor以及后续版本重构后的StreamInputProcessor中。框架收到一个输入的水位时不会立刻向下游转发而是先更新该输入的水位再判断两个输入的水位值把较小的那个作为当前算子水位继续广播。这就是“水位对齐”。4.2 检查点屏障对齐如何保证双流一致性双流 join 的一致性问题除了水位还有检查点。当触发一次检查点时每个输入通道都会收到一个 barrier。如果两个输入通道的 barrier 到达时间不一致算子必须等两个通道的 barrier 都到齐才能做状态快照。这个机制叫“屏障对齐”。它在双流操作中的影响非常直接如果其中一个输入源突然断电或者产生严重反压barrier 迟迟不来另一个输入即便数据已经处理完了也只能阻塞等待整个任务推进速度会被拖慢。我在实际运维中见过一个案例某双流 join 作业的其中一个 Kafka topic 分区数很少消费吞吐跟不上另一个流导致检查点经常超时。表面看是检查点失败实际根因是两条流处理速率差距太大barrier 对齐时间过长。后来把慢流的并行度和分区数提上来检查点就恢复正常了。所以如果你发现双流作业的检查点经常超时不要只盯着检查点配置也要去看两个输入源的处理速率是否匹配。这是 Flink 双流操作源码给我们的一个重要提示。4.3 空闲输入源对双流的影响与应对另一个常见问题是“空闲流”。假设左流一直在来数据右流突然没有新数据了。由于水位对齐机制右流的水位停留在原来的位置整个算子的最小水位就一直不涨。结果就是所有基于事件时间的定时器全部卡住interval join 的清理逻辑不执行窗口 join 不触发。Flink 从 1.11 开始对水源的空闲状态做了优化。如果某个 source 设置了withIdleness(...)连续一段时间没有数据时这个 source 会被标记为“空闲”在计算对齐水位时会被排除掉。也就是说空闲输入不会拖住另一个输入的进度。这个机制在源码层面是StreamStatus控制的。source 发出的数据流中会夹杂IDLE状态标记双流算子感知到后会把该输入从水位对齐逻辑中剔除。如果你用 Kafka source配置assignTimestampsAndWatermarks的时候一定要考虑加上withIdleness。特别是在双流 join 场景里某个业务流夜间几乎没数据是非常正常的不加空闲检测整个作业的定时器都会在夜间卡死第二天早上数据恢复后水位才会重新推进。5. 常见线上问题与排查技巧实录5.1 场景一左右流速率差异巨大双流 join 最常见的性能问题就是两条流的速率不均衡。比如订单流每秒 100 条支付流每秒 10000 条。由于同一个 key 的处理是串行的慢流一条数据要匹配很多快流数据算子的 CPU 压力会集中到某些 key 上。源码层面看KeyedCoProcessOperator处理每条记录的时间取决于用户函数逻辑以及对应状态的大小。如果某个 key 积累了海量未匹配记录每来一条新记录都要遍历整个状态找匹配耗时自然上涨。最终就会表现为整个 subtask 处理耗时增加背压传导到 source。遇到这种情况我一般先查 key 的分布是不是少数大 key 占用了绝大多数数据如果是就需要在业务 key 后面加随机盐做打散分两步 join最后再去掉盐。这个过程会牺牲一些精确性需要业务配合调整。还有一种情况是两条流本身数据量差异就不大但连接器的吞吐能力不匹配。比如一个流来自高并发 Kafka 队列另一个流来自数据库 CDC。此时应该优先做流式缓存预热比如将 CDC 流做成维表广播流而不是直接用双流connect硬碰硬。5.2 场景二状态越来越大状态膨胀几乎是每个双流 join 作业都会遇上的问题。最常见的原因是清理定时器没生效或者清理逻辑写得不对。如果你是用intervalJoin理论上算子会自动清理状态。但要注意它依赖事件时间水位推进。如果水位不推进状态照样膨胀。这时候关键是查 source 的水位生成逻辑。如果你是自己写CoProcessFunction那就更常见了。很多人把数据写进MapState之后忘了在onTimer里删除导致状态只增不减。我的做法是每次put状态之前先计算这条数据什么时候过期然后注册一个事件时间定时器在onTimer里除了处理业务逻辑一定顺手把对应的MapState条目删除。另外可以用状态 TTL 兜底。比如StateTtlConfig设置ttl(Time.hours(1))即使定时器有漏网之鱼过期状态也会被延迟清理。但要留意Flink 的 TTL 不是精确到秒的准时清理它是惰性清理加后台清理结合所以不能作为唯一的清理手段。5.3 场景三join 结果延迟巨大双流 join 结果延迟大通常不是算子本身慢而是水位在等数据。双流水位取最小值如果某条流的事件时间整体落后另一条流即使实时性很好也要等慢的流。排查时第一件事是看 Web UI 里两个输入源的 Current Watermark。如果两个水位差很大问题一定出在慢流的水位生成上。要么是慢流的 source 没有定义正确的事件时间字段要么是读取的 Kafka 分区数据积压。还有一种隐蔽情况某条流的并行子任务中有一个 subtask 收到数据的最大时间戳明显晚于其他 subtask。水位是每个并行通道单独推进的双流算子拿到的水位是所有上游通道水位的最小值任何一个通道掉队都会拖慢整体。解决思路是在上游做分区重平衡让数据时间戳尽量均匀分布到各 subtask。如果业务上允许也可以在 source 端设置withIdleness把个别长时间没数据的通道标记为空闲避免它们拖住整体水位。5.4 定位方法日志加 Web UI 加火焰图排查双流问题我有一套固定的流程。先看 Web UI 的 BackPressure判断哪个算子背压最高。双流 join 的算子通常显示为两个输入背压面板里可以看到两个输入的繁忙率。然后看 Watermark 面板。双流算子会分别显示两个输入水位的当前值。如果两个值长期不接近基本可以锁定是水位对齐问题。再看 Checkpoint 耗时。如果检查点时间明显过长多半是状态太大或者 barrier 等待时间过长。这时候打开作业的 Subtask 时间线看哪个 subtask 的 checkpoint 耗时异常。最后还有一个很有用的手段在CoProcessFunction里临时加日志打印每条记录进入时的时间戳、当前水位、当前状态大小。上线跑十分钟就撤掉。这种“临时探针”往往能最快暴露问题比直接猜要高效得多。如果还定位不了就要上火焰图看算子 CPU 热点但火焰图属于比较高级的排查手段一般到不了这一步。下面是我整理的常见问题速查表现象常见根因排查方向常用解法join 结果延迟大双流水位差大Web UI 看两个输入 watermarks修慢流水位、加空闲检测检查点超时屏障对齐等待看 subtask checkpoint 耗时提升慢流消费能力、调整并行度状态持续膨胀清理定时器未触发看当前水位是否推进修水位、补 onTimer 清理逻辑单个 subtask 高负载key 数据倾斜看 keyBy 分布加盐打散、两阶段 join匹配率低between 参数理解错误核对时间比较方向明确基准流再写上下界6. 双流操作选型心得与一条优化建议6.1 什么时候用 connect什么时候用 intervalJoin很多人问我“到底用connect还是intervalJoin” 我的判断标准很简单。如果你只需要按 key 和时间范围做关联且对匹配率要求高、又能接受事件时间语义优先用intervalJoin。它内部自动管理缓存和清理代码量少也不容易写出状态爆炸的 bug。如果你需要做的是更复杂的流式处理比如“左流记录到达后要等待右流多个事件再做累加或状态机流转”那就用connect CoProcessFunction。你可以完全控制状态结构、定时器逻辑和输出时机灵活度最高。还有一种常见判断intervalJoin适合“两条流都不缺数据只是要按时间窗匹配”connect适合“需要把其中一条流当成动态配置或触发信号来处理”。前者是数据关联后者是状态驱动。6.2 个人建议的源码阅读路径如果你真的想把 flink 双流操作源码啃明白我的建议顺序是先读ConnectedStreams搞清楚 API 层哪些方法是语法糖哪些方法真正生成算子。然后读TwoInputStreamOperator接口建立两个输入的基本心智模型。接着读KeyedCoProcessOperator和CoProcessFunction掌握用户函数是怎么被框架调用的、状态生命周期是怎么管理的。最后读IntervalJoinOperator重点看它如何用ListState加定时器实现缓存和清理。读完这四个类你已经超越了大多数只停留在 API 层的开发者。这时候再回头看线上遇到的任何双流问题脑子里会自动浮现出对应的源码位置排查效率会高很多。最后分享一个我自己的习惯每次写完双流作业我都会在代码注释里把“基准流是谁、时间范围怎么算、状态何时清理”这三件事写清楚。看似多余但半年后再看代码能救你一命。
)
