Flink实现往Kafka中多个topic发送消息

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");

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

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) {
    	// 根据自身的逻辑条件
        JsonUtils.setObjectMapper(new ObjectMapper());
        if("1".equals(flinkJobBO.getApiModel())){
        	// 动态生成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");

// 创建Flink Stream执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
// 1、设置默认topic
String TOPIC = "TEST";
// 2. 从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");
// 从kafka中消费数据
DataStreamSource<String> kafkaDataStream =
        env.addSource(new FlinkKafkaConsumer<>(TOPIC, new SimpleStringSchema(), props));
// 3. 针对流做处理 把string转成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();