做官网网站哪家公司好,嘉兴中小企业网站制作,建设网站过水,手机网站建设制作背景
flink在实现本地内存和db同步配置表信息时#xff0c;想要做到类似于增量(保证实时性) 全量(保证和DB数据一致)的效果#xff0c;那么我们如何通过flink的广播状态外部定时器定时全量同步的方式来实现呢#xff1f;
实现增量全量的效果
package wikiedits.schedule…背景
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 KeyedBroadcastProcessFunctionString, String, String, String {// 键值分区状态private final MapStateDescriptorString, ListString mapStateDesc new MapStateDescriptor(items, BasicTypeInfo.STRING_TYPE_INFO, new ListTypeInfo(String.class));// 广播状态private final MapStateDescriptorString, String ruleStateDescriptor new MapStateDescriptor(RulesBroadcastState, BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO);Overridepublic void processBroadcastElement(String value, Context ctx, CollectorString out) throws Exception {// 1.增量消息更新广播状态BroadcastStateString, String broadcastState ctx.getBroadcastState(ruleStateDescriptor);broadcastState.put(value, value);// 2.全量更新,判断广播状态和DB配置表在本地缓存的配置项是否一致比如如果广播状态记录少了使用本地缓存中的记录来更新下广播状态for (Map.EntryString, 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, CollectorString out) throws Exception {// 键值分区状态final MapStateString, ListString state getRuntimeContext().getMapState(mapStateDesc);// 广播状态for (Map.EntryString, 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 CacheString, String configCache CacheBuilder.newBuilder().initialCapacity(50).maximumSize(500).build();// 通过定时执行器定时同步本地缓存和DB配置表static {scheduledExecutorService Executors.newScheduledThreadPool(10);scheduledExecutorService.scheduleWithFixedDelay(() - {// 2.1 定时任务更新本地内存配置项// ListConfigEntity 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 CacheString, String getConfigCache() {return configCache;}}
总结
1.在处理广播元素的时候除了更新广播状态之外还要对比下广播状态和DB配置表在flink的本地缓存的数据如果不一致需要打印告警日志或者采取更新等措施
2.由于全量广播状态和DB配置表在flink的本地缓存的数据对比是在接收到某个广播元素的时候才进行所以我们可以多余多发送一些相同的广播元素来触发对比
3.通过这种方式广播状态就可以实现增量(实时性) 全量(准确性) 的结果
