使用场景：
在处理数据的时候，有些配置是要实时动态改变的,比如说我要过滤一些关键字，这些关键字呢是在MYSQL里随时配置修改的，那我们在高吞吐计算的Function中动态查询配置文件有可能使整个计算阻塞，甚至任务停止。
广播流可以通过查询配置文件，广播到某个 operator 的所有并发实例中，然后与另一条流数据连接进行计算。
实现步骤：
1、定义一个MapStateDescriptor来描述我们要广播的数据的格式
final MapStateDescriptor<String, String> CONFIG_DESCRIPTOR = new MapStateDescriptor<>(
"wordsConfig",
BasicTypeInfo.STRING_TYPE_INFO,
BasicTypeInfo.STRING_TYPE_INFO);
2、需要一个Stream来广播下游的operator
我这里实现了一个只有1个并发度的数据源，定时查配置文件，发动到下游
public class MinuteBroadcastSource extends RichParallelSourceFunction<String> {
private volatile boolean isRun;
private volatile int lastUpdateMin = -1;
private R2mClusterClient redisDao;
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
isRun = true;
}
@Override
public void run(SourceContext<String> ctx) throws Exception {
while(isRun){
LocalDateTime date = LocalDateTime.now();
int min = date.getMinute();
if(min != lastUpdateMin){
lastUpdateMin = min;
Set<String> configs = readConfigs();
if(configs != null && configs.size() > 0){
for(String config : configs){
ctx.collect(config);
}
}
}
Thread.sleep(1000);
}
}
private Set<String> readConfigs(){
//这里读取配置信息
return null;
}
@Override
public void cancel() {
isRun = false;
}
}
3、添加数据源并把数据源注册成广播流
BroadcastStream<String> broadcastStream = env.addSource(new MinuteBroadcastSource()).setParallelism(1).broadcast(CONFIG_DESCRIPTOR);
4、连接广播流和处理数据的流
DataStream<SkuOrder> connectedStream = orderStream.connect(broadcastStream).process(new BroadcastProcessFunction<Order, String, Order>(){
@Override
public void processElement(Order order, ReadOnlyContext ctx, Collector<Order> collector) throws Exception {
HeapBroadcastState<String,String> config = (HeapBroadcastState)ctx.getBroadcastState(CONFIG_DESCRIPTOR);
Iterator<Map.Entry<String, String>> iterator = config.iterator();
while (iterator.hasNext()){
Map.Entry<String, String> entry =iterator.next();
logger.info("all config:" + entry.getKey() + " value:" + entry.getValue());
}
collector.collect(order);
}
@Override
public void processBroadcastElement(String s, Context ctx, Collector<SkuOrder> collector) throws Exception {
logger.info("收到广播:" + s);
BroadcastState<String,String> state = ctx.getBroadcastState(CONFIG_DESCRIPTOR);
ctx.getBroadcastState(CONFIG_DESCRIPTOR).put(s,"1");
}
});
需要注意到的问题：
1、数据源发送数据时候如果数据是集合，必须使用线程安全的集合类
2、如果上面的MinuteBroadcastSource并行度大于1，那么每一个JOB都会发一条广播，这样的话如果每个JOB一分钟发一次，那么processBroadcastElement会收到 并行度数 * n条消息
3、获取到的BroadcastState是一个map,相同的KEY,put进去会覆盖掉
