flink动态sink到kafka不同的broker和topic中

在 Flink 中，可以使用 `FlinkKafkaProducer` 来将数据写入 Kafka 中。如果要动态地将数据写入到不同的 Kafka 主题或 Broker 中，可以通过构建 `FlinkKafkaProducer` 实例的方式来实现。

首先，要定义一个 `KafkaSerializationSchema` 实现，该实现负责将数据序列化为 Kafka 记录。可以在序列化过程中根据数据的内容动态地确定目标主题和 Broker。

以下是一个简单的示例代码，它根据输入数据的前缀动态地将数据写入到不同的主题和 Broker 中：

import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class DynamicKafkaSink<T> extends RichSinkFunction<T> {

    private transient FlinkKafkaProducer<T> producer;
    private final StringSerializer keySerializer = new StringSerializer();
    private final KafkaSerializationSchema<T> valueSerializer;

    public DynamicKafkaSink(KafkaSerializationSchema<T> valueSerializer) {
        this.valueSerializer = valueSerializer;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);

        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "localhost:9092");
        props.setProperty("acks", "all");

        producer = new FlinkKafkaProducer<>("dummy-topic", valueSerializer, props, FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
    }

    @Override
    public void invoke(T value, Context context) throws Exception {
        String key = getKey(value);

        String topic = getTargetTopic(value);
        String broker = getTargetBroker(value);

        producer.setWriteTimestamp(context.timestamp());
        producer.setTopic(topic);
        producer.setTransactionalId(null);
        producer.setTransactionalState(null);
        producer.setSemantic(FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
        producer.setDefaultTopic(topic);
        producer.setDefaultPartition(null);

        producer.initializeState();
        producer.writeRecord(new ProducerRecord<>(topic, null, context.timestamp(), key, value));

        producer.flush();
    }

    private String getKey(T value) {
        // 根据数据的前缀生成 Key
        String prefix = value.toString().split(":")[0];
        return "key-" + prefix;
    }

    private String getTargetTopic(T value) {
        // 根据数据的前缀选择目标主题
        String prefix = value.toString().split(":")[0];
        return prefix + "-topic";
    }

    private String getTargetBroker(T value) {
        // 根据数据的前缀选择目标 Broker
        String prefix = value.toString().split(":")[0];
        return prefix.equals("a") ? "localhost:9092" : "localhost:9093";
    }
}

在 `invoke` 方法中，根据输入数据的内容确定目标主题和 Broker，并为 `FlinkKafkaProducer` 实例设置相应的属性。然后，将数据写入到目标主题中。

可以在 Flink 应用程序中使用该 `DynamicKafkaSink` 来将数据写入到不同的主题和 Broker 中。例如：

DataStream<String> input = ...;

input.addSink(new DynamicKafkaSink<>(new KafkaSerializationSchema<String>() {
    @Override
    public ProducerRecord<byte[], byte[]> serialize(String element, Long timestamp) {
        String[] parts = element.split(":");
        return new ProducerRecord<>(null, null, parts[0], parts[1].getBytes());
    }
}));

在这个示例中，`DynamicKafkaSink` 实例的泛型类型为 `String`，因此需要提供一个 `KafkaSerializationSchema` 实现来将字符串序列化为 Kafka 记录。在 `serialize` 方法中，使用数据的前缀作为 Key，将数据的后半部分作为 Value，构建一个 Kafka 记录。然后，数据会被动态地写入到不同的主题和 Broker 中。