Flink广播状态的哪种使用方式更好？

Flink 支持两种方式使用广播状态：

1. 直接在 RichFunction 中使用广播状态

public class BroadcastFunction extends RichMapFunction<IN, OUT> {

    private transient MapState<String, String> broadcastState;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 在 open 方法中获取广播状态
        broadcastState = getRuntimeContext().getMapState(
                new MapStateDescriptor<>("broadcast", BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO));
    }

    @Override
    public OUT map(IN value) throws Exception {
        // 在 map 方法中使用广播状态
        String broadcastValue = broadcastState.get("key");
        // ...
    }

    @Override
    public void processBroadcastElement(BROADCAST value, Context ctx, Collector<OUT> out) throws Exception {
        // 在 processBroadcastElement 方法中更新广播状态
        broadcastState.put("key", value.getValue());
    }
}

2. 使用 BroadcastProcessFunction 和 BroadcastStream 进行广播

public class BroadcastFunction extends BroadcastProcessFunction<IN1, IN2, OUT> {

    private transient ValueState<String> valueState;

    @Override
    public void processBroadcastElement(IN2 value, Context ctx, Collector<OUT> out) throws Exception {
        // 在 processBroadcastElement 方法中更新广播状态
        valueState.update(value.getValue());
    }

    @Override
    public void processElement(IN1 value, ReadOnlyContext ctx, Collector<OUT> out) throws Exception {
        // 在 processElement 方法中使用广播状态
        String broadcastValue = valueState.value();
        // ...
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        // 在 open 方法中获取广播状态
        valueState = getRuntimeContext().getState(
                new ValueStateDescriptor<>("broadcast", BasicTypeInfo.STRING_TYPE_INFO));
    }
}

// 将广播流广播出去
BroadcastStream<BROADCAST> broadcastStream = env.addSource(new BroadcastSourceFunction<BROADCAST>());
DataStream<IN1> inStream = ... ;
DataStream<OUT> outStream = inStream.connect(broadcastStream)
        .process(new BroadcastFunction());
``