---

提示：以下是本篇文章正文内容，下面案例可供参考

一、Broadcast

为了关联一个非广播流（keyed 或者 non-keyed）与一个广播流（BroadcastStream），我们可以调用非广播流的方法 connect()，并将 BroadcastStream 当做参数传入。 这个方法的返回参数是 BroadcastConnectedStream，具有类型方法 process()，传入一个特殊的 CoProcessFunction 来书写我们的模式识别逻辑。 具体传入 process() 的是哪个类型取决于非广播流的类型：

• 如果流是一个 keyed 流，那就是 KeyedBroadcastProcessFunction 类型；
• 如果流是一个 non-keyed 流，那就是 BroadcastProcessFunction 类型。

1).例如非keyby的要实现两个方法

public abstract class BroadcastProcessFunction<IN1, IN2, OUT> extends BaseBroadcastProcessFunction {//主流 public abstract void processElement(IN1 value, ReadOnlyContext ctx, Collector<OUT> out) throws Exception;//广播操作public abstract void processBroadcastElement(IN2 value, Context ctx, Collector<OUT> out) throws Exception;
}

2).keyby的

public abstract class KeyedBroadcastProcessFunction<KS, IN1, IN2, OUT> {//主流public abstract void processElement(IN1 value, ReadOnlyContext ctx, Collector<OUT> out) throws Exception;//广播public abstract void processBroadcastElement(IN2 value, Context ctx, Collector<OUT> out) throws Exception;//只有keyby的可以onTimer。此方法可以不重写public void onTimer(long timestamp, OnTimerContext ctx, Collector<OUT> out) throws Exception;
}

在处理广播流元素这端，是具有读写权限的，而对于处理非广播流元素这端是只读的。 这样做的原因是，Flink 中是不存在跨 task 通讯的。所以为了保证 broadcast state 在所有的并发实例中是一致的，我们在处理广播流元素的时候给予写权限，在所有的 task 中均可以看到这些元素，并且要求对这些元素处理是一致的， 那么最终所有 task 得到的 broadcast state 是一致的。
广播算子是不使用 RocksDB state backend: broadcast state 在运行时保存在内存中，需要保证内存充足。这一特性同样适用于所有其他 Operator State。

二、代码示例

此处将本地csv文件加载到内存广播中
CSV文件的内容是:
1.user_details.csv

1,Alice,30
2,Bob,25

2.user_details03.csv

3,Charlie,35
5,name,5

下面是代码(下面是将两个本地CSV文件放到广播内存中案例)

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;import java.util.HashMap;
import java.util.Map;
public <