快捷導(dǎo)航

Flink實(shí)現(xiàn)往Kafka中多個(gè)topic發(fā)送消息

更新時(shí)間：2025年10月13日 09:44:47 作者：暴走的Aluuubbarrrr

文章介紹了使用Flink 1.13.2 和 Kafka 2.6.2 從Kafka讀取數(shù)據(jù),并根據(jù)邏輯將數(shù)據(jù)分配到不同的topic,提到了需要重寫FlinkKafka的Key序列化器,并加入自定義邏輯以發(fā)送消息到指定的topic,文中還包括了配置Kafka信息和如何連接FlinkKafka的步驟

思路與環(huán)境

從kafka中讀取數(shù)據(jù) 根據(jù)邏輯判斷分配到不同的topic中去

需要重寫Flink Kafka的Key序列化器，并通過(guò)加入自己的邏輯主動(dòng)往指定的topic發(fā)送消息。

首先配置Kafka信息

Properties props = new Properties();
props.put("bootstrap.servers","10.116.0.16:9092");
props.put("acks", "all");
props.put("retries", 1);
props.put("batch.size", 16384);
props.put("linger.ms", 1);
props.put("buffer.memory", 33554432);
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

注意這里key序列化器和value序列化器都為StringSerializer

Flink Kafka連接器

FlinkKafkaProducer<FlinkJobBO> fkProducer =
                new FlinkKafkaProducer<>("", new MyKeySerialization(), props, FlinkKafkaProducer.Semantic.EXACTLY_ONCE);

其中MyKeySerialization便是重寫的key序列化器

自定義序列化器

public class MyKeySerialization implements KafkaSerializationSchema<FlinkJobBO> {
    String topic;
    public MyKeySerialization(String topic){
        this.topic = topic;
    }
    public MyKeySerialization(){
    }

	// 注意：都是byte[]類型，所以我們要重新指定新的序列化器
    @Override
    public ProducerRecord<byte[], byte[]> serialize(FlinkJobBO flinkJobBO, @Nullable Long aLong) {
    	// 根據(jù)自身的邏輯條件
        JsonUtils.setObjectMapper(new ObjectMapper());
        if("1".equals(flinkJobBO.getApiModel())){
        	// 動(dòng)態(tài)生成topic
            return new ProducerRecord<>("topic-"+flinkJobBO.getGroupId(), JsonUtils.toJson(flinkJobBO).getBytes(StandardCharsets.UTF_8));
        }
        return new ProducerRecord<>("", "".getBytes(StandardCharsets.UTF_8));
    }
}

需要把

props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

替換為

props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");

完整代碼

// 創(chuàng)建Flink Stream執(zhí)行環(huán)境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
// 1、設(shè)置默認(rèn)topic
String TOPIC = "TEST";
// 2. 從kafka獲取流數(shù)據(jù)
Properties props = new Properties();
props.put("bootstrap.servers","10.116.0.16:9092");
props.put("acks", "all");
props.put("retries", 1);
props.put("batch.size", 16384);
props.put("linger.ms", 1);
props.put("buffer.memory", 33554432);
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
// 從kafka中消費(fèi)數(shù)據(jù)
DataStreamSource<String> kafkaDataStream =
        env.addSource(new FlinkKafkaConsumer<>(TOPIC, new SimpleStringSchema(), props));
// 3. 針對(duì)流做處理 把string轉(zhuǎn)成bo 主流
DataStream<FlinkJobBO> ds = kafkaDataStream
        .map((MapFunction<String, FlinkJobBO>) s -> JsonUtils.toBean(s, FlinkJobBO.class));
// 修改value序列化器
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
// 4.1.1 自定義序列化器 分配topic *****
FlinkKafkaProducer<FlinkJobBO> fkProducer =
        new FlinkKafkaProducer<>("", new MyKeySerialization(), props, FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
fkProducer.setLogFailuresOnly(false);
ds.addSink(fkProducer);
env.execute();