我没有尝试过这种方法，但我相信，与其使用setTopic将接收器硬连接到特定主题，不如在自定义KafkaRecordSerializationSchema上实现serialize方法，以便它返回的每个ProducerRecord都指定它应该写入的主题。
另一种选择是为每个主题创建一个单独的接收器对象，然后使用一个ProcessFunction，该ProcessFunction扇出到一组端输出，每个端输出连接到适当的接收器。

我可以通过使用@DavidAnderson建议的自定义序列化方法实现KafkaRecordSerializationSchema，将输出接收到多个Kafka主题。

public class CustomSchema implements KafkaRecordSerializationSchema<Tuple2<String,String>> {

private final String encoding = StandardCharsets.UTF_8.name();

@Override
public ProducerRecord<byte[], byte[]> serialize(Tuple2<String, String> input, KafkaSinkContext kafkaSinkContext, Long aLong) {
    String topic = input.f0;
    String data = input.f1;
    try {
        byte[] value = data==null ? null:data.getBytes(this.encoding);
        return new ProducerRecord<>(topic,value);
    } catch (UnsupportedEncodingException e) {
        throw new SerializationException("Error when serializing string to byte[] due to unsupported encoding " + this.encoding);
    }
}

我通过setRecordSerializer方法配置了Kafka接收器来使用它。