当前位置：首页 > news >正文

flink以增量+全量的方式更新广播状态

news 2025/7/12 10:07:57

背景

flink在实现本地内存和db同步配置表信息时，想要做到类似于增量(保证实时性) + 全量(保证和DB数据一致)的效果，那么我们如何通过flink的广播状态+外部定时器定时全量同步的方式来实现呢？

实现增量+全量的效果

package wikiedits.schedule;import java.util.List;
import java.util.Map;import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.typeutils.ListTypeInfo;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;//处理函数
public class BroadcastStatePlusSchedulerFunction extends KeyedBroadcastProcessFunction<String, String, String, String> {// 键值分区状态private final MapStateDescriptor<String, List<String>> mapStateDesc =new MapStateDescriptor<>("items", BasicTypeInfo.STRING_TYPE_INFO, new ListTypeInfo<>(String.class));// 广播状态private final MapStateDescriptor<String, String> ruleStateDescriptor = new MapStateDescriptor<>("RulesBroadcastState", BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO);@Overridepublic void processBroadcastElement(String value, Context ctx, Collector<String> out) throws Exception {// 1.增量消息更新广播状态BroadcastState<String, String> broadcastState = ctx.getBroadcastState(ruleStateDescriptor);broadcastState.put(value, value);// 2.全量更新,判断广播状态和DB配置表在本地缓存的配置项是否一致，比如如果广播状态记录少了，使用本地缓存中的记录来更新下广播状态for (Map.Entry<String, String> entry : StaticLoadUtil.getConfigCache().asMap().entrySet()) {String broadcastValue = broadcastState.get(entry.getKey());if(!StringUtils.equals(entry.getValue(), broadcastValue)){//如果不相等,那么以DB缓存中的为准}}// 3.自此，广播状态和DB配置表的状态几乎一致,不过由于他们的比较只发生于收到广播元素，所以我们可以在凌晨的时候故意从db中找出几条记录发送kafka消息到这个广播状态来进行触发比较,当然这里也可以当收到某个元素时覆盖掉flink的广播状态}@Overridepublic void processElement(String value, ReadOnlyContext ctx, Collector<String> out) throws Exception {// 键值分区状态final MapState<String, List<String>> state = getRuntimeContext().getMapState(mapStateDesc);// 广播状态for (Map.Entry<String, String> entry : ctx.getBroadcastState(ruleStateDescriptor).immutableEntries()) {}}}// 外部定时器实现
package wikiedits.schedule;import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;/*** 静态类定时加载DB配置表到本地内存中*/
public class StaticLoadUtil {// 定时任务执行器private static transient ScheduledExecutorService scheduledExecutorService;public static final Cache<String, String> configCache =CacheBuilder.newBuilder().initialCapacity(50).maximumSize(500).build();// 通过定时执行器定时同步本地缓存和DB配置表static {scheduledExecutorService = Executors.newScheduledThreadPool(10);scheduledExecutorService.scheduleWithFixedDelay(() -> {// 2.1 定时任务更新本地内存配置项// List<ConfigEntity> configList = DBManager.SELECTSQL.getConfigs();// for(ConfigEntity entity : configList){configCache.put("key", "value");// }// 2.2 更新本地变量threshold的值// threshold = DBManager.SELECTSQL.getConfig("threshold");}, 0, 100, TimeUnit.SECONDS);}/*** 获取本地缓存*/public static Cache<String, String> getConfigCache() {return configCache;}}