A read-only view of the {@link BroadcastState}.
*
* <p>Although read-only, the user code should not modify the value returned by
the {@link
* #get(Object)} or the entries of the immutable iterator returned by the {@link
* #immutableEntries()}, as this can lead to inconsistent states. The reason for
this is that we do
* not create extra copies of the elements for performance reasons.
*
* @param <K> The key type of the elements in the {@link ReadOnlyBroadcastState}.
* @param <V> The value type of the elements in the {@link
ReadOnlyBroadcastState}.
*/
这是源码中对ReadOnlyBroadcastState的描述，希望对你有帮助
smq
发件人: chaos
发送时间: 2021年3月10日 14:29
收件人: user-zh@flink.apache.org
主题: Re: 回复：MapState 无法更新问题
主要代码如下：
class getRule extends KeyedBroadcastProcessFunction[String,
KafkaStreamSource, List[Rule], KafkaStreamSource] {
private var carEfenceState: MapState[String, Boolean] = _
override def open(parameters: Configuration): Unit = {
carEfenceState = getRuntimeContext.getMapState(new
MapStateDescriptor[String, Boolean]("carEfenceState", classOf[String],
classOf[Boolean]))
}
override def processBroadcastElement(in2: List[Rule], context:
KeyedBroadcastProcessFunction[String, KafkaStreamSource, List[Rule],
KafkaStreamSource]#Context, collector: Collector[KafkaStreamSource]): Unit =
{
context.getBroadcastState(ruleStateDescriptor).put("rules", in2)
}
override def processElement(kafkaSource: KafkaStreamSource,
readOnlyContext: KeyedBroadcastProcessFunction[String, KafkaStreamSource,
List[Rule], KafkaStreamSource]#ReadOnlyContext, collector:
Collector[KafkaStreamSource]): Unit = {
val ruleIterator =
readOnlyContext.getBroadcastState(ruleStateDescriptor).immutableEntries().iterator()
while (ruleIterator.hasNext) {
val ruleMap: Map.Entry[String, List[Rule]] =
ruleIterator.next()
val ruleList: List[Rule] = ruleMap.getValue
for (rule <- ruleList) {
val mapKey = kafkaSource.vno + rule.id
val tempState = carEfenceState.get(mapKey)
val currentState = if (tempState != null)
tempState else false
// 业务逻辑
if (!currentState) {
...
carEfenceState.put(mapKey, true)
...
} else if (currentState) {
...
carEfenceState.remove(mapKey)
...
}
}
}
}
}
--
Sent from: http://apache-flink.147419.n8.nabble.com/
c
