flink写入kafka默认只写到一个partition

如题使用flink FlinkKafkaProducer0* 时，有一个最常用，也是官方例子的构造方法 有三个参数如下

/**
* @param topicId
*        ID of the Kafka topic.
* @param serializationSchema
*        User defined key-less serialization schema.
* @param producerConfig
*        Properties with the producer configuration.
*/
public FlinkKafkaProducer010(String topicId, SerializationSchema<T> serializationSchema, Properties producerConfig) {
   this(topicId, new KeyedSerializationSchemaWrapper<>(serializationSchema), producerConfig, new FlinkFixedPartitioner<T>());
}

运行一段时间后发现所有的数据都写到一个kafka的partition了，这样就不能发挥kafka的并行能力。于是去阅读官方文档发现

官方文档中描述大概意思是

“默认情况下，如果没有为Flink Kafka生成器指定自定义分区器，那么该生成器将使用一个FlinkFixedPartitioner将每个Flink Kafka生成器的并行子任务映射到单个Kafka分区(即，接收子任务接收到的所有记录最终将位于相同的Kafka分区中)。如果没有指定自定义分区程序，则使用FlinkFixedPartitioner。“

如果kafka的partition和和flink 算子的并行度刚好数量一致则没问题，不然就会出现上诉问题。

于是找到找到一个重载方法设置FlinkFixedPartitioner。

new FlinkKafkaProducer010<>(topic, new SimpleStringSchema(), config, (FlinkKafkaPartitioner)null);

public FlinkKafkaProducer010(
      String topicId,
      SerializationSchema<T> serializationSchema,
      Properties producerConfig,
      @Nullable FlinkKafkaPartitioner<T> customPartitioner) {

   this(topicId, new KeyedSerializationSchemaWrapper<>(serializationSchema), producerConfig, customPartitioner);
}

有兴趣的同学可以跟到源码里面去看看flink 是怎么样分配数据去哪个partition的

FlinkFixedPartitioner