@Test def testRestoreSavepointWithBroadcast(): Unit = {
 val env = StreamExecutionEnvironment.getExecutionEnvironment env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
 // 以下两个变量是为了确定广播流发出的数据类型，广播流可以同时发出多种类型的数据 lazy val firstBroadcastStateDesc = new MapStateDescriptor[Long, Long]( "broadcast-state-1", BasicTypeInfo.LONG_TYPE_INFO.asInstanceOf[TypeInformation[Long]], BasicTypeInfo.LONG_TYPE_INFO.asInstanceOf[TypeInformation[Long]])
 lazy val secondBroadcastStateDesc = new MapStateDescriptor[String, String]( "broadcast-state-2", BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO)
 env.setStateBackend(new MemoryStateBackend) env.enableCheckpointing(500) env.setParallelism(4) env.setMaxParallelism(4)
 // 数据流，这里数据流和广播流的Source都是同一种CheckpointedSource。数据流这里做了一系列算子操作，比如flatMap val stream = env .addSource( new CheckpointedSource(4)).setMaxParallelism(1).uid("checkpointedSource") .keyBy( new KeySelector[(Long, Long), Long] { override def getKey(value: (Long, Long)): Long = value._1 } ) .flatMap(new StatefulFlatMapper) .keyBy( new KeySelector[(Long, Long), Long] { override def getKey(value: (Long, Long)): Long = value._1 } )
 // 广播流 val broadcastStream = env .addSource( new CheckpointedSource(4)).setMaxParallelism(1).uid("checkpointedBroadcastSource") .broadcast(firstBroadcastStateDesc, secondBroadcastStateDesc)
 // 把数据流和广播流结合起来 stream .connect(broadcastStream) .process(new VerifyingBroadcastProcessFunction(expectedFirstState, expectedSecondState)) .addSink(new AccumulatorCountingSink) }}
// 用户自定义的处理函数class TestBroadcastProcessFunction extends KeyedBroadcastProcessFunction [Long, (Long, Long), (Long, Long), (Long, Long)] {
 // 重点说明，这里的 firstBroadcastStateDesc，secondBroadcastStateDesc 其实和之前广播流的那两个MapStateDescriptor无关。  // 这里两个MapStateDescriptor是为了存取BroadcastState，这样在 processBroadcastElement和processElement之间就可以传递变量了。我们完全可以定义新的MapStateDescriptor，只要processBroadcastElement和processElement之间认可就行。  // 这里参数 "broadcast-state-1" 是name, flink就是用这个 name 来从Flink运行时系统中存取MapStateDescriptor  lazy val firstBroadcastStateDesc = new MapStateDescriptor[Long, Long]( "broadcast-state-1", BasicTypeInfo.LONG_TYPE_INFO.asInstanceOf[TypeInformation[Long]], BasicTypeInfo.LONG_TYPE_INFO.asInstanceOf[TypeInformation[Long]])
 val secondBroadcastStateDesc = new MapStateDescriptor[String, String]( "broadcast-state-2", BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO)
 override def processElement( value: (Long, Long), ctx: KeyedBroadcastProcessFunction [Long, (Long, Long), (Long, Long), (Long, Long)]#ReadOnlyContext, out: Collector[(Long, Long)]): Unit = {
 // 这里Flink源码中是直接把接受到的业务变量直接再次转发出去 out.collect(value)  }
 override def processBroadcastElement( value: (Long, Long), ctx: KeyedBroadcastProcessFunction [Long, (Long, Long), (Long, Long), (Long, Long)]#Context, out: Collector[(Long, Long)]): Unit = { // 这里是把最新传来的广播变量存储起来，processElement中可以取出再次使用. 具体是通过firstBroadcastStateDesc 的 name 来获取 BroadcastState ctx.getBroadcastState(firstBroadcastStateDesc).put(value._1, value._2) ctx.getBroadcastState(secondBroadcastStateDesc).put(value._1.toString, value._2.toString) }}
// 广播流和数据流的Sourceprivate class CheckpointedSource(val numElements: Int) extends SourceFunction[(Long, Long)] with CheckpointedFunction {
 private var isRunning = true private var state: ListState[CustomCaseClass] = _
 // 就是简单的定期发送 override def run(ctx: SourceFunction.SourceContext[(Long, Long)]) { ctx.emitWatermark(new Watermark(0)) ctx.getCheckpointLock synchronized { var i = 0 while (i < numElements) { ctx.collect(i, i) i += 1 } } // don't emit a final watermark so that we don't trigger the registered event-time // timers while (isRunning) Thread.sleep(20) }}

// 因为主要起到控制作用，所以这个函数的处理相对简单override def processBroadcastElement(): Unit = { // 这里可以把最新传来的广播变量存储起来，processElement中可以取出再次使用，比如 ctx.getBroadcastState(firstBroadcastStateDesc).put(value._1, value._2)}

// 这个函数需要和processBroadcastElement配合起来使用override def processElement(): Unit = { // 可以取出processBroadcastElement之前存储的广播变量，然后用此来处理业务变量，比如 val secondBroadcastStateDesc = new MapStateDescriptor[String, String]( "broadcast-state-2", BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO) 
 var actualSecondState = Map[String, String]() for (entry <- ctx.getBroadcastState(secondBroadcastStateDesc).immutableEntries()) { val v = secondExpectedBroadcastState.get(entry.getKey).get actualSecondState += (entry.getKey -> entry.getValue) } 
 // 甚至这里只要和processBroadcastElement一起关联好，可以存储任意类型的变量。不必须要和广播变量的类型一致。重点是声明新的对应的MapStateDescriptor // MapStateDescriptor继承了StateDescriptor，其中state为MapState类型，value为Map类型}

// 模式始终存储在MapState中，并将null作为键。broadcast state始终表示为MapState，这是Flink提供的最通用的状态原语。MapStateDescriptor<Void, Pattern> bcStateDescriptor =  new MapStateDescriptor<>("patterns", Types.VOID, Types.POJO(Pattern.class));
// 能看到的是，在处理广播变量时候，存储广播变量到BroadcastState public void processBroadcastElement(Pattern pattern, Context ctx,  Collector<Tuple2<Long, Pattern>> out) throws Exception { // store the new pattern by updating the broadcast state BroadcastState<Void, Pattern> bcState = ctx.getBroadcastState(patternDesc); // storing in MapState with null as VOID default value bcState.put(null, pattern); }
// 能看到的是，在处理业务变量时候，从BroadcastState取出广播变量，存取时候实际都是用"patterns"这个name字符串来作为key。 public void processElement(Action action, ReadOnlyContext ctx,  Collector<Tuple2<Long, Pattern>> out) throws Exception { // get current pattern from broadcast state Pattern pattern = ctx.getBroadcastState(this.patternDesc) // access MapState with null as VOID default value .get(null); // get previous action of current user from keyed state String prevAction = prevActionState.value(); if (pattern != null && prevAction != null) { // user had an action before, check if pattern matches if (pattern.firstAction.equals(prevAction) &&  pattern.secondAction.equals(action.action)) { // MATCH out.collect(new Tuple2<>(ctx.getCurrentKey(), pattern)); } } // update keyed state and remember action for next pattern evaluation prevActionState.update(action.action); }

 * The life cycle of the Broadcast: * {@code * -- 初始化逻辑 -> 用一个BroadcastConnectedStream把数据流和广播流结合起来进行拓扑转换 * |  * +----> businessStream = DataStream.filter.map.... * | // 处理业务逻辑的数据流，businessStream 是普通DataStream  * +----> broadcastStream = DataStream.broadcast(broadcastStateDesc) * | // 处理配置逻辑的广播数据流，broadcastStream是BroadcastStream类型 * +----> businessStream.connect(broadcastStream) * | .process(new processFunction(broadcastStateDesc)) * | // 把业务流,广播流 结合起来，生成一个BroadcastConnectedStream，然后进行 process * +----------> process @ BroadcastConnectedStream  * | TwoInputStreamOperator<IN1, IN2, OUT> operator = * | new CoBroadcastWithNonKeyedOperator<>(clean(function), * | broadcastStateDescriptors); * | return transform(outTypeInfo, operator);  * | // 生成一个类型是TwoInputStreamOperator 的 operator，进行 transform * +----------------> transform @ BroadcastConnectedStream  * | transform = new TwoInputTransformation<>( * | inputStream1.getTransformation(), // 业务流 * | inputStream2.getTransformation(), // 广播流 * | ifunctionName, // 用户的UDF * | operator, // 算子 CoBroadcastWithNonKeyedOperator * | outTypeInfo); // 输出类型 * | returnStream = new SingleOutputStreamOperator(transform); * | getExecutionEnvironment().addOperator(transform) * | // 将业务流,广播流与拓扑联合起来形成一个转换，加到 Env 中，这就完成了拓扑转换  * | // 最后返回结果是一个SingleOutputStreamOperator。 * }

 * 数据结构: * -- BroadcastStream.  * 就是简单封装一个DataStream，然后记录这个广播流对应的StateDescriptors  public class BroadcastStream<T> {  private final StreamExecutionEnvironment environment; private final DataStream<T> inputStream; private final List<MapStateDescriptor<?, ?>> broadcastStateDescriptors;  }  * 数据结构: * -- BroadcastConnectedStream.  * 把业务流,广播流 结合起来，然后会生成算子和拓扑public class BroadcastConnectedStream<IN1, IN2> { private final StreamExecutionEnvironment environment; private final DataStream<IN1> inputStream1; private final BroadcastStream<IN2> inputStream2; private final List<MapStateDescriptor<?, ?>> broadcastStateDescriptors;} 
* 真实计算:* -- CoBroadcastWithNonKeyedOperator -> 真正对BroadcastProcessFunction的执行，是在这里完成的public class CoBroadcastWithNonKeyedOperator<IN1, IN2, OUT> extends AbstractUdfStreamOperator<OUT, BroadcastProcessFunction<IN1, IN2, OUT>> implements TwoInputStreamOperator<IN1, IN2, OUT> {  private final List<MapStateDescriptor<?, ?>> broadcastStateDescriptors; private transient TimestampedCollector<OUT> collector; private transient Map<MapStateDescriptor<?, ?>, BroadcastState<?, ?>> broadcastStates; private transient ReadWriteContextImpl rwContext; private transient ReadOnlyContextImpl rContext;  @Override public void processElement1(StreamRecord<IN1> element) throws Exception { collector.setTimestamp(element); rContext.setElement(element); // 当上游有最新业务数据来的时候，调用用户自定义的processElement // 在这可以把之前存储的广播配置信息取出，然后对业务数据流进行处理  userFunction.processElement(element.getValue(), rContext, collector); rContext.setElement(null); }
 @Override public void processElement2(StreamRecord<IN2> element) throws Exception { collector.setTimestamp(element); rwContext.setElement(element); // 当上游有数据来的时候，调用用户自定义的processBroadcastElement // 在这可以把最新传送的广播配置信息存起来  userFunction.processBroadcastElement(element.getValue(), rwContext, collector); rwContext.setElement(null); } }

// 就是connect，broadcast，分别生成对应的数据流public class DataStream<T> { protected final StreamExecutionEnvironment environment; protected final Transformation<T> transformation;
 @PublicEvolving public <R> BroadcastConnectedStream<T, R> connect(BroadcastStream<R> broadcastStream) { return new BroadcastConnectedStream<>( environment, this, Preconditions.checkNotNull(broadcastStream), broadcastStream.getBroadcastStateDescriptor()); }  @PublicEvolving public BroadcastStream<T> broadcast(final MapStateDescriptor<?, ?>... broadcastStateDescriptors) { final DataStream<T> broadcastStream = setConnectionType(new BroadcastPartitioner<>()); return new BroadcastStream<>(environment, broadcastStream, broadcastStateDescriptors); }}

public class MapStateDescriptor<UK, UV> extends StateDescriptor<MapState<UK, UV>, Map<UK, UV>> { /** * Create a new {@code MapStateDescriptor} with the given name and the given type serializers. * * @param name The name of the {@code MapStateDescriptor}. * @param keySerializer The type serializer for the keys in the state. * @param valueSerializer The type serializer for the values in the state. */  public MapStateDescriptor(String name, TypeSerializer<UK> keySerializer, TypeSerializer<UV> valueSerializer) { super(name, new MapSerializer<>(keySerializer, valueSerializer), null); }
 /** * Create a new {@code MapStateDescriptor} with the given name and the given type information. * * @param name The name of the {@code MapStateDescriptor}. * @param keyTypeInfo The type information for the keys in the state. * @param valueTypeInfo The type information for the values in the state. */ public MapStateDescriptor(String name, TypeInformation<UK> keyTypeInfo, TypeInformation<UV> valueTypeInfo) { super(name, new MapTypeInfo<>(keyTypeInfo, valueTypeInfo), null); }
 /** * Create a new {@code MapStateDescriptor} with the given name and the given type information. * * <p>If this constructor fails (because it is not possible to describe the type via a class), * consider using the {@link #MapStateDescriptor(String, TypeInformation, TypeInformation)} constructor. * * @param name The name of the {@code MapStateDescriptor}. * @param keyClass The class of the type of keys in the state. * @param valueClass The class of the type of values in the state. */ public MapStateDescriptor(String name, Class<UK> keyClass, Class<UV> valueClass) { super(name, new MapTypeInfo<>(keyClass, valueClass), null); }}

public class DefaultOperatorStateBackend implements OperatorStateBackend {  /** * Map for all registered operator states. Maps state name -> state */ private final Map<String, PartitionableListState<?>> registeredOperatorStates;
 /** * Map for all registered operator broadcast states. Maps state name -> state */ private final Map<String, BackendWritableBroadcastState<?, ?>> registeredBroadcastStates;   /** * Cache of already accessed states. * * <p>In contrast to {@link #registeredOperatorStates} which may be repopulated * with restored state, this map is always empty at the beginning. * * <p>TODO this map should be moved to a base class once we have proper hierarchy for the operator state backends. */ private final Map<String, PartitionableListState<?>> accessedStatesByName;
 private final Map<String, BackendWritableBroadcastState<?, ?>> accessedBroadcastStatesByName; // 这里用来缓存广播变量  // 这里就是前文中所说的，存取广播变量的API public <K, V> BroadcastState<K, V> getBroadcastState(final MapStateDescriptor<K, V> stateDescriptor) throws StateMigrationException {
 String name = Preconditions.checkNotNull(stateDescriptor.getName());
 // 如果之前有，就取出来 BackendWritableBroadcastState<K, V> previous = (BackendWritableBroadcastState<K, V>) accessedBroadcastStatesByName.get( name);
 if (previous != null) { return previous; }
 stateDescriptor.initializeSerializerUnlessSet(getExecutionConfig()); TypeSerializer<K> broadcastStateKeySerializer = Preconditions.checkNotNull(stateDescriptor.getKeySerializer()); TypeSerializer<V> broadcastStateValueSerializer = Preconditions.checkNotNull(stateDescriptor.getValueSerializer());
 BackendWritableBroadcastState<K, V> broadcastState = (BackendWritableBroadcastState<K, V>) registeredBroadcastStates.get(name);
 if (broadcastState == null) { broadcastState = new HeapBroadcastState<>( new RegisteredBroadcastStateBackendMetaInfo<>( name, OperatorStateHandle.Mode.BROADCAST, broadcastStateKeySerializer, broadcastStateValueSerializer)); registeredBroadcastStates.put(name, broadcastState); } else { // has restored state; check compatibility of new state access
 RegisteredBroadcastStateBackendMetaInfo<K, V> restoredBroadcastStateMetaInfo = broadcastState.getStateMetaInfo();
 // check whether new serializers are incompatible TypeSerializerSchemaCompatibility<K> keyCompatibility = restoredBroadcastStateMetaInfo.updateKeySerializer(broadcastStateKeySerializer);
 TypeSerializerSchemaCompatibility<V> valueCompatibility = restoredBroadcastStateMetaInfo.updateValueSerializer(broadcastStateValueSerializer);
 broadcastState.setStateMetaInfo(restoredBroadcastStateMetaInfo); }
 accessedBroadcastStatesByName.put(name, broadcastState); // 如果之前没有，就存入 return broadcastState; } }