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

背景

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);



    @Override
    public 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的广播状态
    }

    @Override
    public 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;
    }


}